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()}
|