# -*- coding: utf-8 -*- """看板命令行入口。 python app.py init # 建库 / 迁移 / 写入黑名单种子（离线） python app.py sync # 收藏夹只读同步（首次导入或增量） python app.py snapshot # 采集当日七项快照与粉丝数 python app.py discover # 执行一次搜索发现并写入候选池 python app.py candidates # 查看候选池（离线） python app.py auth-check # 检查 Cookie 登录态 python app.py run # 按配置周期常驻运行（--once 只跑一次） python app.py status # 查看库内状态（离线） P2 起 ``python app.py`` 将在此基础上启动本地看板服务。 """ from __future__ import annotations import argparse import json import sys import threading import time from pathlib import Path from typing import Sequence from .bilibili.auth_flow import AuthService from .bilibili.client import BilibiliClient, Pacer from .bilibili.provider import ClientProvider from .bilibili.search import SearchClient from .bilibili.write_client import BilibiliWriteClient from .config import ConfigError, ConfigStore from .database import Database from .logging_json import JsonLogger from .notify.telegram import TelegramClient from .repository import Repository from .scheduler import ScheduledTask, Scheduler from .services.discovery import DiscoveryService from .services.favorites import FavoriteActionService from .services.folder_sync import FolderSyncService from .services.snapshot import SnapshotService from .services.telegram_poller import TelegramPoller from .services.up_profile import UpProfileService from .timeutil import next_daily_run, parse_hhmm, shanghai_date ROOT = Path(__file__).resolve().parent.parent DEFAULT_CONFIG = ROOT / "config.json" def _build_parser() -> argparse.ArgumentParser: common = argparse.ArgumentParser(add_help=False) common.add_argument( "--config", default=argparse.SUPPRESS, help="配置文件路径（默认项目根目录 config.json）", ) parser = argparse.ArgumentParser( prog="app.py", description="咻咻满 Reaction 数据看板 —— 数据底座、同步与每日快照", ) parser.add_argument( "--config", default=str(DEFAULT_CONFIG), help="配置文件路径（默认项目根目录 config.json）", ) sub = parser.add_subparsers(dest="command", required=False) sub.add_parser("init", parents=[common], help="建库、迁移并写入黑名单种子") sync = sub.add_parser("sync", parents=[common], help="同步已配置的年度收藏夹") sync.add_argument( "--full", action="store_true", help="强制全量刷新元数据（忽略 6 小时周期）" ) sync.add_argument( "--allow-empty-folder", action="store_true", help="允许把远端 0 条视为真实清空（默认按异常跳过差集）", ) sync.add_argument( "--no-capture-followers", action="store_true", help="跳过同步末尾的 UP 粉丝采集（改用 backfill-followers 单独回填）", ) snapshot = sub.add_parser( "snapshot", parents=[common], help="采集当日七项快照与粉丝数" ) snapshot.add_argument( "--force", action="store_true", help="忽略当日完成标记，重新采集" ) discover = sub.add_parser( "discover", parents=[common], help="执行一次搜索发现并写入候选池" ) discover.add_argument( "--no-notify", action="store_true", help="只写候选池，不发送 Telegram 通知", ) candidates = sub.add_parser( "candidates", parents=[common], help="查看候选池（离线）" ) candidates.add_argument( "--decision", choices=("accepted", "rejected"), default=None, help="只查看指定决策状态的候选", ) auth = sub.add_parser("auth", parents=[common], help="扫码授权并更新指定账号 Cookie") auth.add_argument("--account", choices=("read", "write"), default="read", help="要授权的账号角色（默认 read）") auth.add_argument("--no-browser", action="store_true", help="不自动打开二维码图片") auth.add_argument("--timeout", type=int, default=180, metavar="SEC", help="等待扫码的超时时间（默认 180 秒）") auth_check = sub.add_parser("auth-check", parents=[common], help="检查账号登录态") auth_check.add_argument("--account", choices=("read", "write", "all"), default="read", help="要检查的账号角色（默认 read）") run = sub.add_parser("run", parents=[common], help="按配置周期常驻运行同步与快照") run.add_argument( "--once", action="store_true", help="只执行一次到期任务后退出" ) run.add_argument( "--allow-empty-folder", action="store_true", help="允许把远端 0 条视为真实清空（默认按异常跳过差集）", ) serve = sub.add_parser("serve", parents=[common], help="启动只读本地看板（不运行调度）") serve.add_argument( "--preview", action="store_true", help="以预览模式启动（要求数据库含 preview:seeded=1 标记）", ) backfill = sub.add_parser( "backfill-followers", parents=[common], help="为历史视频回填 UP 粉丝数快照（只补空值）", ) backfill.add_argument("--limit", type=int, default=None, help="最多处理多少个 UP") backfill.add_argument("--dry-run", action="store_true", help="只统计待回填数量，不发请求") sub.add_parser("status", parents=[common], help="查看数据库状态（离线）") return parser def _open_db(cfg, logger: JsonLogger) -> Database: db = Database( cfg.storage.database_path, backup_dir=cfg.storage.backup_dir, backup_keep_days=cfg.storage.backup_keep_days, logger=logger, ) db.initialize() return db def _print(payload) -> None: print(json.dumps(payload, ensure_ascii=False, indent=2), flush=True) def _record_task( repo: Repository, name: str, *, ok: bool, started_at: int | None = None, finished_at: int | None = None, payload=None, error: str | None = None, ) -> None: """把最近一次任务结果写入 kv，供只读系统页展示。""" now = int(time.time()) record = { "name": name, "ok": bool(ok), "started_at": int(started_at if started_at is not None else now), "finished_at": int(finished_at if finished_at is not None else now), "error": error, "summary": payload, } try: repo.kv_set(f"task:last:{name}", json.dumps(record, ensure_ascii=False)) except Exception: # noqa: BLE001 任务状态记录失败不影响主流程 pass def _client(credentials, logger: JsonLogger, *, pacer=None, retries: int = 3) -> BilibiliClient: return BilibiliClient( cookie=credentials.cookie, csrf=credentials.csrf, mid=credentials.mid, pacer=pacer, retries=retries, logger=logger, ) def cmd_init(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) for folder in cfg.folders: repo.ensure_folder(folder) seeded = repo.seed_blacklist(cfg.up_blacklist_seed) _print({ "ok": True, "schema_version": db.schema_version, "database": str(cfg.storage.database_path), "blacklist_seeded": seeded, **repo.counts(), }) finally: db.close() return 0 def cmd_sync(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=True) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) client = _client(cfg.bilibili.read, logger) service = FolderSyncService( store, repo, client, logger=logger, allow_empty_folders=args.allow_empty_folder, capture_followers=not getattr(args, "no_capture_followers", False), ) started = int(time.time()) summary = service.run(full_refresh=args.full) payload = summary.as_dict() payload["database"] = str(cfg.storage.database_path) payload["counts"] = repo.counts() _record_task( repo, "folder_sync", ok=summary.ok, started_at=started, finished_at=int(time.time()), payload=payload, ) _print(payload) return 0 if summary.ok else 1 finally: db.close() def cmd_backfill_followers(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) client = _client( cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=2 ) service = UpProfileService(repo, lambda: client, logger=logger) started = int(time.time()) result = service.backfill(limit=args.limit, dry_run=args.dry_run) payload = result.as_dict() payload["dry_run"] = bool(args.dry_run) payload["coverage"] = repo.up_follower_coverage() payload["database"] = str(cfg.storage.database_path) _record_task( repo, "backfill_followers", ok=result.failed == 0, started_at=started, finished_at=int(time.time()), payload=payload, error="；".join(result.errors) if result.errors else None, ) _print(payload) return 0 if result.failed == 0 else 1 finally: db.close() def cmd_snapshot(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) client = _client( cfg.bilibili.read, logger, pacer=Pacer(1.0, 1.6), retries=1 ) service = SnapshotService(store, repo, client, logger=logger) summary = service.run_if_needed(force=args.force) payload = summary.as_dict() payload["database"] = str(cfg.storage.database_path) payload["snapshot_counts"] = repo.snapshot_counts(summary.stat_date) payload["counts"] = repo.counts() ok = bool(summary.completed and not summary.backup_error) _record_task( repo, "snapshot", ok=ok, started_at=int(summary.started_at or time.time()), finished_at=int(time.time()), payload=payload, error=summary.backup_error, ) _print(payload) return 0 if ok else 1 finally: db.close() def _telegram(cfg, logger: JsonLogger): if not cfg.telegram.enabled: return None if not cfg.telegram.bot_token or not cfg.telegram.chat_id: raise ConfigError( "telegram.enabled=true 时必须配置 bot_token 和 chat_id" ) return TelegramClient( cfg.telegram.bot_token, cfg.telegram.chat_id, logger=logger, ) def cmd_discover(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=True) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) provider = ClientProvider( store, "read", lambda credentials: _client(credentials, logger) ) service = DiscoveryService( store, repo, SearchClient(provider.get), notifier=None if args.no_notify else _telegram(cfg, logger), logger=logger, ) summary = service.run() payload = summary.as_dict() payload["candidate_total"] = len(repo.list_candidates()) payload["database"] = str(cfg.storage.database_path) _record_task( repo, "discovery", ok=not summary.errors, started_at=int(summary.started_at), finished_at=int(summary.finished_at or time.time()), payload=payload, error="；".join(summary.errors) if summary.errors else None, ) _print(payload) return 0 if not summary.errors else 1 finally: db.close() def cmd_candidates(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) rows = repo.list_candidates(args.decision) _print({ "ok": True, "count": len(rows), "candidates": rows, "database": str(cfg.storage.database_path), }) return 0 finally: db.close() def cmd_auth(args, logger: JsonLogger) -> int: if args.timeout < 1: raise ConfigError("auth --timeout 必须至少为 1 秒") return AuthService(Path(args.config), account=args.account, logger=logger).run( timeout=args.timeout, no_browser=args.no_browser ) def _auth_check(cfg, account: str, logger: JsonLogger) -> dict[str, object]: credentials = getattr(cfg.bilibili, account) if not credentials.cookie: return {"account": account, "ok": False, "is_login": False, "message": "未配置 cookie"} state = _client(credentials, logger, retries=1).check_login() return { "account": account, "ok": state.is_login, "is_login": state.is_login, "mid": state.mid, "uname": state.uname, "code": state.code, "message": state.message, } def cmd_auth_check(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() if args.account != "all": result = _auth_check(cfg, args.account, logger) _print(result) return 0 if result["ok"] else 1 results = [_auth_check(cfg, account, logger) for account in ("read", "write")] _print({"ok": all(result["ok"] for result in results), "accounts": results}) return 0 if all(result["ok"] for result in results) else 1 def cmd_run(args, logger: JsonLogger, *, serve_web: bool = False) -> int: store = ConfigStore(args.config, require_auth=True) cfg = store.get() db = _open_db(cfg, logger) last: dict[str, object] = {} web_server = None try: repo = Repository(db) folder_provider = ClientProvider( store, "read", lambda credentials: _client(credentials, logger) ) snapshot_provider = ClientProvider( store, "read", lambda credentials: _client( credentials, logger, pacer=Pacer(1.0, 1.6), retries=1 ), ) folder_service = FolderSyncService( store, repo, client_provider=folder_provider, logger=logger, allow_empty_folders=args.allow_empty_folder, ) snapshot_service = SnapshotService( store, repo, client_provider=snapshot_provider, logger=logger ) write_provider = ClientProvider( store, "write", lambda credentials: BilibiliWriteClient( cookie=credentials.cookie, csrf=credentials.csrf, mid=credentials.mid, logger=logger, ), ) favorite_service = FavoriteActionService( store, repo, write_provider, logger=logger ) notifier = _telegram(cfg, logger) discovery_service = DiscoveryService( store, repo, SearchClient(folder_provider.get), notifier=notifier, logger=logger, ) def run_discover(): started = int(time.time()) try: summary = discovery_service.run() except Exception as exc: # noqa: BLE001 记录后交给调度器隔离 _record_task( repo, "discovery", ok=False, started_at=started, finished_at=int(time.time()), error=str(exc), ) raise payload = summary.as_dict() _record_task( repo, "discovery", ok=not summary.errors, started_at=summary.started_at, finished_at=summary.finished_at or int(time.time()), payload=payload, error="；".join(summary.errors) if summary.errors else None, ) last["discover"] = summary logger.info("discovery.summary", **payload) return summary def run_cleanup(): started = int(time.time()) removed = repo.delete_expired_candidates( ttl_days=cfg.schedule.candidate_ttl_days ) expired = repo.expire_fav_actions() if removed or expired: repo.bump_cache_version("videos") logger.info( "cleanup.done", candidates_removed=removed, fav_actions_expired=expired, ) _record_task( repo, "candidate_cleanup", ok=True, started_at=started, finished_at=int(time.time()), payload={ "candidates_removed": removed, "fav_actions_expired": expired, }, ) return { "candidates_removed": removed, "fav_actions_expired": expired, } def run_snapshot(): started = int(time.time()) try: summary = snapshot_service.run_if_needed() except Exception as exc: # noqa: BLE001 _record_task( repo, "snapshot", ok=False, started_at=started, finished_at=int(time.time()), error=str(exc), ) raise payload = summary.as_dict() ok = bool(summary.completed and not summary.backup_error) _record_task( repo, "snapshot", ok=ok, started_at=summary.started_at, finished_at=summary.finished_at or int(time.time()), payload=payload, error=summary.backup_error, ) last["snapshot"] = summary logger.info("snapshot.summary", **payload) return summary def run_sync(): started = int(time.time()) try: summary = folder_service.run() except Exception as exc: # noqa: BLE001 _record_task( repo, "folder_sync", ok=False, started_at=started, finished_at=int(time.time()), error=str(exc), ) raise payload = summary.as_dict() _record_task( repo, "folder_sync", ok=summary.ok, started_at=started, finished_at=int(time.time()), payload=payload, error=None if summary.ok else "；".join(summary.errors), ) last["folder"] = summary logger.info("folder_sync.summary", **payload) # 新入库/恢复的视频立即补当天快照，而不是等到次日。 if summary.added > 0: run_snapshot() return summary def run_config_watch(): before = store.revision store.get() if store.revision != before: logger.info("config.reloaded", revision=store.revision) return {"revision": store.revision} hh, mm = parse_hhmm(cfg.schedule.snapshot_at) snapshot_at = f"{hh:02d}:{mm:02d}" scheduler = Scheduler( [ ScheduledTask( "folder_sync", float(cfg.schedule.folder_sync_minutes * 60), run_sync, ), ScheduledTask( "snapshot", 86400.0, run_snapshot, next_run=0.0, next_run_fn=lambda started: next_daily_run(started, snapshot_at), ), ScheduledTask( "discovery", float(cfg.schedule.discover_minutes * 60), run_discover, ), ScheduledTask("candidate_cleanup", 86400.0, run_cleanup), ScheduledTask("config_watch", 60.0, run_config_watch), ], logger=logger, ) if serve_web and not args.once: from .web.server import create_server, serve_forever web_server = create_server(store, db, logger=logger) threading.Thread( target=lambda: serve_forever(web_server, logger=logger), name="web-server", daemon=True, ).start() if notifier is not None and not args.once: poller = TelegramPoller(notifier, favorite_service, logger=logger) threading.Thread( target=poller.run_forever, name="telegram-poller", daemon=True, ).start() logger.info("telegram.polling_started") if args.once: results = scheduler.run_due() _print([result.as_dict() for result in results]) else: logger.info( "scheduler.start", folder_sync_seconds=cfg.schedule.folder_sync_minutes * 60, snapshot_at=snapshot_at, ) scheduler.run_forever() folder_summary = last.get("folder") snapshot_summary = last.get("snapshot") discover_summary = last.get("discover") folder_ok = folder_summary is None or getattr(folder_summary, "ok", True) snapshot_ok = snapshot_summary is None or getattr(snapshot_summary, "completed", True) discover_ok = discover_summary is None or not getattr(discover_summary, "errors", []) return 0 if folder_ok and snapshot_ok and discover_ok else 1 finally: if web_server is not None: try: web_server.shutdown() finally: web_server.server_close() db.close() def _assert_preview_database(path: Path) -> None: import sqlite3 if not path.exists(): raise ConfigError(f"预览数据库不存在: {path}") uri = f"file:{path.resolve().as_posix()}?mode=ro" try: conn = sqlite3.connect(uri, uri=True) except sqlite3.Error as exc: raise ConfigError(f"无法读取预览数据库: {exc}") from exc try: row = conn.execute( "SELECT v FROM kv WHERE k = 'preview:seeded'" ).fetchone() except sqlite3.Error as exc: raise ConfigError(f"目标数据库不是预览库: {exc}") from exc finally: conn.close() if row is None or str(row[0]) != "1": raise ConfigError( "目标数据库缺少 preview:seeded=1 标记；请先运行 " "tools/seed_preview_db.py，或去掉 --preview" ) def cmd_serve(args, logger: JsonLogger) -> int: from .web.server import create_server, serve_forever store = ConfigStore(args.config, require_auth=False) cfg = store.get() if args.preview: _assert_preview_database(cfg.storage.database_path) db = _open_db(cfg, logger) try: server = create_server(store, db, preview=args.preview, logger=logger) serve_forever(server, logger=logger) return 0 finally: db.close() def cmd_status(args, logger: JsonLogger) -> int: store = ConfigStore(args.config, require_auth=False) cfg = store.get() db = _open_db(cfg, logger) try: repo = Repository(db) today = shanghai_date() snapshot = repo.snapshot_counts(today) snapshot.update({ "date": today, "last_attempt": repo.kv_get("snapshot:last_attempt"), "last_completed": repo.kv_get("snapshot:last_completed"), "auth_ok": repo.kv_get("auth:ok"), "auth_message": repo.kv_get("auth:message"), "backup_ok": repo.kv_get(f"snapshot:backup:{today}"), "backup_error": repo.kv_get(f"snapshot:backup_error:{today}"), }) _print({ "ok": True, "schema_version": db.schema_version, "database": str(cfg.storage.database_path), "configured_folders": [ { "media_id": row.media_id, "year": row.year, "title": row.title, "media_count": row.media_count, "last_ids_at": row.last_ids_at, "last_full_at": row.last_full_at, "enabled": row.enabled, } for row in repo.list_folders() ], "snapshot": snapshot, **repo.counts(), }) finally: db.close() return 0 def main(argv: Sequence[str] | None = None) -> int: args = _build_parser().parse_args(list(argv) if argv is not None else None) logger = JsonLogger() try: if args.command is None: # 无子命令时启动完整服务：只读 Web + 调度 + Telegram。 args.command = "run" args.once = False args.allow_empty_folder = False return cmd_run(args, logger, serve_web=True) if args.command == "serve": return cmd_serve(args, logger) if args.command == "init": return cmd_init(args, logger) if args.command == "sync": return cmd_sync(args, logger) if args.command == "snapshot": return cmd_snapshot(args, logger) if args.command == "backfill-followers": return cmd_backfill_followers(args, logger) if args.command == "discover": return cmd_discover(args, logger) if args.command == "candidates": return cmd_candidates(args, logger) if args.command == "auth": return cmd_auth(args, logger) if args.command == "auth-check": return cmd_auth_check(args, logger) if args.command == "run": return cmd_run(args, logger) if args.command == "status": return cmd_status(args, logger) except ConfigError as exc: logger.error("config.error", error=str(exc)) return 2 except KeyboardInterrupt: logger.warning("app.interrupted") return 130 except Exception as exc: # noqa: BLE001 CLI 边界统一记录 logger.error("app.error", error=str(exc)) return 2 logger.error("app.unknown_command", command=getattr(args, "command", None)) return 2 if __name__ == "__main__": sys.exit(main())