104 lines
3.2 KiB
Python
104 lines
3.2 KiB
Python
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 "<html" not in html.lower() and "<!doctype html" not in html.lower():
|
||
raise TaskError("RULE_HTML_INVALID", "请上传完整 HTML 文档", 400)
|
||
|
||
_ensure_rule_dir()
|
||
updated_at = now_iso()
|
||
RULE_HTML_PATH.write_text(html, encoding="utf-8")
|
||
RULE_META_PATH.write_text(
|
||
json_dumps({
|
||
"title": title,
|
||
"filename": filename,
|
||
"updated_at": updated_at,
|
||
}),
|
||
encoding="utf-8",
|
||
)
|
||
return get_rule_document()
|
||
|
||
|
||
def delete_rule_document() -> 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()}
|