110 lines
4.3 KiB
Python
Executable File
110 lines
4.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
import requests
|
|
|
|
SKILL_DIR = Path(__file__).resolve().parent.parent
|
|
WORKTREE = Path('/Users/karuo/Documents/卡若创业派对工作树')
|
|
API = 'https://soulapi.quwanzhi.com'
|
|
STATE = SKILL_DIR / 'state.json'
|
|
EXPORT_ROOT = SKILL_DIR / 'exports'
|
|
|
|
|
|
def load_module(name: str, path: Path):
|
|
import importlib.util
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def latest_chapter() -> tuple[int, str, str]:
|
|
parts = requests.get(f'{API}/api/miniprogram/book/parts', timeout=30).json().get('parts', [])
|
|
found = []
|
|
for part in parts:
|
|
if not str(part.get('id', '')).startswith('part-2026-'):
|
|
continue
|
|
data = requests.get(f"{API}/api/miniprogram/book/chapters-by-part?partId={part['id']}", timeout=30).json()
|
|
for row in data.get('data', []):
|
|
m = re.search(r'第\s*(\d+)\s*场', row.get('sectionTitle', ''))
|
|
if m and row.get('id'):
|
|
found.append((int(m.group(1)), row['id'], row.get('sectionTitle', '')))
|
|
if not found:
|
|
raise RuntimeError('未找到 2026 派对文章')
|
|
return max(found, key=lambda x: x[0])
|
|
|
|
|
|
def download_article(field: int, out_dir: Path) -> Path:
|
|
cmd = ['python3', str(WORKTREE / 'scripts/content_download.py'), str(field), '--out-dir', str(out_dir), '--base', API]
|
|
result = subprocess.run(cmd, capture_output=True, text=True, check=True)
|
|
match = re.search(r'已写入:\s*(.+)', result.stdout)
|
|
if not match:
|
|
raise RuntimeError('文章下载成功但未定位输出文件')
|
|
return Path(match.group(1).strip())
|
|
|
|
|
|
def download_images(raw: str, out_dir: Path) -> list[Path]:
|
|
urls = re.findall(r'https?://[^\s"\'<>\)]+', raw)
|
|
urls = [u.rstrip('.,;') for u in urls if any(x in u.lower() for x in ('.png', '.jpg', '.jpeg', 'oss', 'image'))]
|
|
result = []
|
|
for i, url in enumerate(dict.fromkeys(urls), 1):
|
|
r = requests.get(url, timeout=60)
|
|
r.raise_for_status()
|
|
suffix = Path(urlparse(url).path).suffix.lower()
|
|
if suffix not in {'.png', '.jpg', '.jpeg'}:
|
|
suffix = '.png'
|
|
path = out_dir / f'图{i}{suffix}'
|
|
path.write_bytes(r.content)
|
|
result.append(path)
|
|
return result
|
|
|
|
|
|
def main() -> int:
|
|
sender = load_module('sender', WORKTREE / 'scripts/send_feishu_text_and_images.py')
|
|
poster = load_module('poster', WORKTREE / 'scripts/send_chapter_poster_to_feishu.py')
|
|
field, chapter_id, title = latest_chapter()
|
|
state = json.loads(STATE.read_text(encoding='utf-8')) if STATE.exists() else {}
|
|
if field <= int(state.get('last_sent_field', 0)):
|
|
print(f'NO_NEW_ARTICLE field={field}')
|
|
return 0
|
|
package = EXPORT_ROOT / f'第{field}场_{datetime.now():%Y%m%d}'
|
|
package.mkdir(parents=True, exist_ok=True)
|
|
article = download_article(field, package)
|
|
raw = article.read_text(encoding='utf-8')
|
|
images = download_images(raw, package)
|
|
body = sender.clean_rich_text(raw)
|
|
final = '关注派对房号 FM29037620\n\n' + body
|
|
clean_file = package / f'第{field}场文章_飞书版.txt'
|
|
clean_file.write_text(final + '\n', encoding='utf-8')
|
|
if re.search(r'<[^>]+>', final):
|
|
raise RuntimeError('正文仍含 HTML 标签')
|
|
if not sender.send_text(poster.WEBHOOK, final[:20000]):
|
|
raise RuntimeError('正文发送失败')
|
|
token = sender.tenant_token()
|
|
if not token:
|
|
raise RuntimeError('获取飞书应用 token 失败')
|
|
for image in images:
|
|
key = sender.upload_png(token, image)
|
|
if not key or not sender.send_image(poster.WEBHOOK, key):
|
|
raise RuntimeError(f'图片发送失败: {image.name}')
|
|
STATE.write_text(json.dumps({'last_sent_field': field, 'chapter_id': chapter_id, 'title': title, 'sent_at': datetime.now().isoformat(timespec='seconds')}, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
|
|
print(f'SENT field={field} images={len(images)} file={clean_file}')
|
|
return 0
|
|
|
|
|
|
if __name__ == '__main__':
|
|
try:
|
|
raise SystemExit(main())
|
|
except Exception as exc:
|
|
print(f'FAILED {type(exc).__name__}: {exc}', file=sys.stderr)
|
|
raise
|