from pathlib import Path from typing import Any from flask import Request from config import Config from database import json_dumps, json_loads, now_iso from task_service import TaskError RULE_DIR = Config.FILE_STORAGE_PATH / "rules" RULE_HTML_PATH = RULE_DIR / "current.html" RULE_META_PATH = RULE_DIR / "current.json" def _ensure_rule_dir() -> None: RULE_DIR.mkdir(parents=True, exist_ok=True) def _read_text(path: Path) -> str: return path.read_text(encoding="utf-8") def get_rule_document() -> dict[str, Any]: """读取当前规则 HTML 的公开元数据。""" if not RULE_HTML_PATH.exists(): return { "exists": False, "title": "", "filename": "", "updated_at": None, "size_bytes": 0, "html_url": "/api/v1/rules/current.html", } metadata = json_loads(_read_text(RULE_META_PATH)) if RULE_META_PATH.exists() else {} stat = RULE_HTML_PATH.stat() return { "exists": True, "title": metadata.get("title") or "小红书问一问收录规则探索", "filename": metadata.get("filename") or RULE_HTML_PATH.name, "updated_at": metadata.get("updated_at"), "size_bytes": stat.st_size, "html_url": "/api/v1/rules/current.html", } def get_rule_html_path() -> Path: if not RULE_HTML_PATH.exists(): raise TaskError("RULE_NOT_FOUND", "当前没有已上传的规则 HTML", 404) return RULE_HTML_PATH def save_rule_document(request: Request) -> dict[str, Any]: """保存当前规则 HTML；支持 multipart 文件、JSON html 字段和 text/html 原文。""" filename = "current.html" title = "小红书问一问收录规则探索" html = "" uploaded = request.files.get("file") if uploaded: filename = uploaded.filename or filename html = uploaded.read().decode("utf-8") title = request.form.get("title") or title else: payload = request.get_json(silent=True) if isinstance(payload, dict): html = str(payload.get("html") or "") filename = str(payload.get("filename") or filename) title = str(payload.get("title") or title) else: html = request.get_data(as_text=True) or "" title = request.args.get("title") or title filename = request.args.get("filename") or filename if not html.strip(): raise TaskError("RULE_HTML_REQUIRED", "html 内容不能为空", 400) if " dict[str, Any]: """删除当前规则 HTML 和元数据。""" deleted = False for path in (RULE_HTML_PATH, RULE_META_PATH): if path.exists(): path.unlink() deleted = True return {"deleted": deleted, **get_rule_document()}