hz4th_coder/material-library
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d682952f41d7393000ac428a9a85458353d80f6a
material-library/app.py
T

822 lines
32 KiB
Python
Raw Normal View History

素材库系统 v1.0.0: 项目/素材管理 + 大模型分析 + FTS全文检索
2026-08-27 12:56:14 +08:00
# -*- coding: utf-8 -*-
"""
素材库系统 (Material Library)
- 项目: 分类/标签/描述管理
- 素材: 上传文件(文字/文档/图片/视频/音频) + 直接粘贴文本
- 文本抽取: txt/md/pdf/docx/pptx/csv/srt/json/xml/html
- 大模型分析: DeepSeek 提取摘要/要点/关键词/标签/归类; 图片走 doubao 视觉
- 索引: SQLite FTS5 (jieba 分词) 全文检索
"""
import os
import re
import io
import json
import uuid
import shutil
import threading
import datetime
import sqlite3
from pathlib import Path
import requests
import jieba
from flask import Flask, request, jsonify, send_file, abort, g
# ---------------- 基础配置 ----------------
BASE_DIR = Path(__file__).resolve().parent
DATA_DIR = BASE_DIR / "data"
UPLOAD_DIR = DATA_DIR / "uploads"
DB_PATH = DATA_DIR / "material.db"
STATIC_DIR = BASE_DIR / "static"
os.makedirs(UPLOAD_DIR, exist_ok=True)
os.makedirs(STATIC_DIR, exist_ok=True)
MAX_FILE_MB = 500 # 单文件上限
MAX_TEXT_LEN = 30000 # 抽取文本送入 LLM 的最大长度
# ---------------- 大模型配置 ----------------
LLM_BASE_URL = os.environ.get("LLM_BASE_URL", "https://api.deepseek.com")
LLM_API_KEY = os.environ.get("LLM_API_KEY", "sk-edb9df58ff574f8c98df1cd6a425e97c")
LLM_MODEL = os.environ.get("LLM_MODEL", "deepseek-v4-flash")
VISION_BASE_URL = os.environ.get("VISION_BASE_URL", "https://ark.cn-beijing.volces.com/api/plan/v3")
VISION_API_KEY = os.environ.get("VISION_API_KEY", "ark-2b06dc9d-8878-4c6e-b201-f422376e79cb-246d9")
VISION_MODEL = os.environ.get("VISION_MODEL", "doubao-seed-evolving")
app = Flask(__name__, static_folder=str(STATIC_DIR), static_url_path="/static")
app.config["MAX_CONTENT_LENGTH"] = MAX_FILE_MB * 1024 * 1024
_lock = threading.RLock()
# 允许的文件扩展名 -> 素材类型
TYPE_BY_EXT = {
".txt": "text", ".md": "text", ".markdown": "text", ".log": "text", ".srt": "text",
".csv": "text", ".json": "text", ".xml": "text", ".html": "text", ".htm": "text",
".pdf": "doc", ".docx": "doc", ".doc": "doc", ".pptx": "doc", ".ppt": "doc",
".jpg": "image", ".jpeg": "image", ".png": "image", ".gif": "image", ".bmp": "image",
".webp": "image", ".tif": "image", ".tiff": "image", ".svg": "image", ".ico": "image",
".mp4": "video", ".mov": "video", ".avi": "video", ".mkv": "video", ".flv": "video",
".wmv": "video", ".webm": "video", ".m4v": "video", ".ts": "video",
".mp3": "audio", ".wav": "audio", ".flac": "audio", ".aac": "audio", ".ogg": "audio",
".m4a": "audio", ".wma": "audio", ".amr": "audio",
}
TEXT_EXTS = {".txt", ".md", ".markdown", ".log", ".srt", ".csv", ".json", ".xml", ".html", ".htm"}
DOC_EXTS = {".pdf", ".docx", ".pptx"}
IMAGE_EXTS = {k for k, v in TYPE_BY_EXT.items() if v == "image"}
VIDEO_EXTS = {k for k, v in TYPE_BY_EXT.items() if v == "video"}
AUDIO_EXTS = {k for k, v in TYPE_BY_EXT.items() if v == "audio"}
ALLOWED_EXT = set(TYPE_BY_EXT.keys())
# ---------------- 数据库 ----------------
def get_db():
if "db" not in g:
g.db = sqlite3.connect(str(DB_PATH))
g.db.row_factory = sqlite3.Row
g.db.execute("PRAGMA journal_mode=WAL")
g.db.execute("PRAGMA foreign_keys=ON")
return g.db
def close_db(e=None):
db = g.pop("db", None)
if db is not None:
db.close()
def init_db():
db = sqlite3.connect(str(DB_PATH))
db.executescript("""
CREATE TABLE IF NOT EXISTS projects (
id INTEGER PRIMARY KEY AUTOINCREMENT,
name TEXT NOT NULL,
description TEXT DEFAULT '',
category TEXT DEFAULT '未分类',
tags TEXT DEFAULT '',
created_at TEXT DEFAULT (datetime('now','localtime')),
updated_at TEXT DEFAULT (datetime('now','localtime'))
);
CREATE TABLE IF NOT EXISTS materials (
id INTEGER PRIMARY KEY AUTOINCREMENT,
project_id INTEGER NOT NULL,
name TEXT NOT NULL,
mtype TEXT DEFAULT 'text',
original_name TEXT DEFAULT '',
file_path TEXT DEFAULT '',
file_size INTEGER DEFAULT 0,
ext TEXT DEFAULT '',
meta_json TEXT DEFAULT '{}',
extracted_text TEXT DEFAULT '',
text_status TEXT DEFAULT 'pending',
status TEXT DEFAULT 'uploaded',
created_at TEXT DEFAULT (datetime('now','localtime'))
);
CREATE TABLE IF NOT EXISTS analyses (
id INTEGER PRIMARY KEY AUTOINCREMENT,
material_id INTEGER,
project_id INTEGER,
scope TEXT DEFAULT 'material',
summary TEXT DEFAULT '',
key_points TEXT DEFAULT '[]',
keywords TEXT DEFAULT '[]',
tags TEXT DEFAULT '[]',
category TEXT DEFAULT '',
detail TEXT DEFAULT '{}',
model TEXT DEFAULT '',
created_at TEXT DEFAULT (datetime('now','localtime'))
);
CREATE INDEX IF NOT EXISTS idx_mat_project ON materials(project_id);
CREATE INDEX IF NOT EXISTS idx_ana_project ON analyses(project_id);
CREATE VIRTUAL TABLE IF NOT EXISTS materials_fts USING fts5(term);
""")
db.commit()
db.close()
def _ts():
return datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def _tokenize(text):
return " ".join(t for t in jieba.cut(text or "") if t.strip())
def _index_material(db, mid, name, text, summary="", keywords=""):
term = " ".join(filter(None, [_tokenize(name), _tokenize(text), _tokenize(summary), _tokenize(keywords)]))
db.execute("DELETE FROM materials_fts WHERE rowid=?", (mid,))
if term.strip():
db.execute("INSERT INTO materials_fts(rowid, term) VALUES (?,?)", (mid, term))
# ---------------- 文本抽取 ----------------
def _read_plain(path):
"""读取纯文本，尝试多种编码"""
raw = path.read_bytes()
for enc in ("utf-8", "utf-8-sig", "gb18030", "big5", "latin-1"):
try:
return raw.decode(enc)
except Exception:
continue
return raw.decode("utf-8", errors="ignore")
def _extract_docx(path):
import docx
d = docx.Document(str(path))
parts = [p.text for p in d.paragraphs if p.text.strip()]
for tbl in d.tables:
for row in tbl.rows:
cells = [c.text.strip() for c in row.cells if c.text.strip()]
if cells:
parts.append(" | ".join(cells))
return "\n".join(parts)
def _extract_pptx(path):
import zipfile
import xml.etree.ElementTree as ET
ns = {"a": "http://schemas.openxmlformats.org/drawingml/2006/main"}
parts = []
with zipfile.ZipFile(str(path)) as z:
slides = sorted([n for n in z.namelist() if re.match(r"ppt/slides/slide\d+\.xml$", n)],
key=lambda x: int(re.search(r"\d+", x.split("/")[-1]).group()))
for s in slides:
try:
root = ET.fromstring(z.read(s))
texts = [t.text for t in root.iter("{http://schemas.openxmlformats.org/drawingml/2006/main}t") if t.text and t.text.strip()]
if texts:
parts.append(" ".join(texts))
except Exception:
continue
return "\n".join(parts)
def _extract_pdf(path):
import pypdf
reader = pypdf.PdfReader(str(path))
parts = []
for i, page in enumerate(reader.pages[:200]):
try:
t = page.extract_text() or ""
if t.strip():
parts.append(t)
except Exception:
continue
return "\n".join(parts)
def extract_text(path, ext):
"""返回 (text, ok)"""
try:
if ext in (".pdf",):
t = _extract_pdf(path)
elif ext == ".docx":
t = _extract_docx(path)
elif ext == ".pptx":
t = _extract_pptx(path)
elif ext in TEXT_EXTS:
t = _read_plain(path)
else:
return "", False
t = re.sub(r"[ \t]+", " ", t)
t = re.sub(r"\n{3,}", "\n\n", t)
return t.strip(), bool(t.strip())
except Exception as e:
return f"<抽取失败: {e}>", False
def _meta_image(path):
from PIL import Image
try:
im = Image.open(str(path))
return {"width": im.width, "height": im.height, "format": im.format or ""}
except Exception:
return {}
def _safe_filename(name):
name = Path(name).name
name = re.sub(r"[\\/:*?\"<>|\s]+", "_", name)
return name[:120] or "unnamed"
# ---------------- 大模型调用 ----------------
def _llm_chat(messages, model=None, temperature=0.3, max_tokens=1500):
url = f"{LLM_BASE_URL.rstrip('/')}/chat/completions"
payload = {
"model": model or LLM_MODEL,
"messages": messages,
"temperature": temperature,
"max_tokens": max_tokens,
"response_format": {"type": "json_object"},
}
r = requests.post(url, headers={"Authorization": f"Bearer {LLM_API_KEY}"},
json=payload, timeout=120)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"]
ANALYZE_PROMPT = """你是专业素材库分析助手。请对下面这个素材做深度分析，提取关键信息。
只输出合法 JSON，不要任何额外文字，格式如下：
{{"summary":"120字以内中文摘要","key_points":["要点1","要点2"...最多6条],"keywords":["关键词1"...最多8个],"tags":["标签1"...最多5个],"category":"建议归类，从[技术文档,产品资料,研究报告,市场资讯,项目管理,图片素材,音视频素材,其他]中选择"}}
素材名称：{name}
素材类型：{mtype}
{extra}
素材内容：
{content}
"""
def _parse_json(content):
try:
return json.loads(content)
except Exception:
m = re.search(r"\{.*\}", content, re.S)
if m:
try:
return json.loads(m.group())
except Exception:
pass
return {}
def _vision_describe(path):
"""用 doubao 视觉模型描述图片"""
try:
import base64
b64 = base64.b64encode(Path(path).read_bytes()).decode()
url = f"{VISION_BASE_URL.rstrip('/')}/chat/completions"
payload = {
"model": VISION_MODEL,
"messages": [{
"role": "user",
"content": [
{"type": "text", "text": "请用中文简洁描述这张图片的内容（80字以内），说明主体、场景、画面中的文字信息（如有）。只输出描述本身。"},
{"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
],
}],
"temperature": 0.3,
"max_tokens": 300,
}
r = requests.post(url, headers={"Authorization": f"Bearer {VISION_API_KEY}"},
json=payload, timeout=120)
r.raise_for_status()
return r.json()["choices"][0]["message"]["content"].strip()
except Exception:
return None
def analyze_material(mat):
"""分析单个素材，返回 (analysis_dict, model, detail)"""
if not isinstance(mat, dict):
mat = dict(mat)
detail = {}
mtype = mat["mtype"]
content = mat.get("extracted_text") or ""
extra = ""
if mtype == "image" and mat.get("file_path"):
desc = _vision_describe(mat["file_path"])
if desc:
extra = "图片视觉描述：\n" + desc
detail["vision"] = desc
if mtype == "video":
extra = "视频素材（未转写音频），基于文件名与元数据分析"
if mtype == "audio":
extra = "音频素材（未转写语音），基于文件名与元数据分析"
if mtype == "other" or not content:
content = content or "（该素材无可提取文本内容）"
text = content[:MAX_TEXT_LEN]
prompt = ANALYZE_PROMPT.format(name=mat["name"], mtype=mtype, extra=extra, content=text)
out = _llm_chat([{"role": "user", "content": prompt}])
data = _parse_json(out)
detail["raw"] = out[:3000]
detail["model"] = LLM_MODEL
return data, LLM_MODEL, detail
def _store_analysis(db, material_id, project_id, scope, data, model, detail):
if scope == "material":
db.execute("DELETE FROM analyses WHERE material_id=? AND scope='material'", (material_id,))
else:
db.execute("DELETE FROM analyses WHERE project_id=? AND scope='project'", (project_id,))
cur = db.execute(
"INSERT INTO analyses (material_id, project_id, scope, summary, key_points, keywords, tags, category, detail, model, created_at) "
"VALUES (?,?,?,?,?,?,?,?,?,?,?)",
(material_id, project_id, scope,
(data.get("summary") or ""),
json.dumps(data.get("key_points", []), ensure_ascii=False),
json.dumps(data.get("keywords", []), ensure_ascii=False),
json.dumps(data.get("tags", []), ensure_ascii=False),
(data.get("category") or ""),
json.dumps(detail, ensure_ascii=False),
model, _ts()))
db.commit()
return cur.lastrowid
# ---------------- 后台任务 ----------------
_tasks = {}
def _task_key(scope, obj_id):
return f"{scope}:{obj_id}"
def _run_analysis(project_id, material_ids=None, scope="material"):
key = _task_key(scope, project_id)
_tasks[key] = {"running": True, "done": 0, "total": 0, "msg": "准备中", "error": None}
db = sqlite3.connect(str(DB_PATH))
db.row_factory = sqlite3.Row
try:
if scope == "project":
cond = "project_id=? AND (status='uploaded' OR status='error' OR status='analyzed')"
mats = db.execute(f"SELECT * FROM materials WHERE {cond}", (project_id,)).fetchall()
else:
if material_ids:
ph = ",".join("?" for _ in material_ids)
mats = db.execute(f"SELECT * FROM materials WHERE project_id=? AND id IN ({ph})", (project_id, *material_ids)).fetchall()
else:
# 未指定 id → 分析项目下所有待分析/失败素材
mats = db.execute(
"SELECT * FROM materials WHERE project_id=? AND (status='uploaded' OR status='error' OR status='analyzed')",
(project_id,)).fetchall()
_tasks[key]["total"] = len(mats)
for i, mat in enumerate(mats, 1):
mid = mat["id"]
try:
db.execute("UPDATE materials SET status='analyzing' WHERE id=?", (mid,))
db.commit()
data, model, detail = analyze_material(mat)
aid = _store_analysis(db, mid, project_id, "material", data, model, detail)
# 回写关键词作为标签候选 + 更新索引
keywords = data.get("keywords", [])
kw = ",".join(keywords[:8]) if isinstance(keywords, list) else ""
db.execute("UPDATE materials SET status='analyzed', name=name WHERE id=?", (mid,))
row = db.execute("SELECT name, extracted_text, summary FROM materials LEFT JOIN analyses ON analyses.id=? WHERE materials.id=?", (aid, mid)).fetchone()
_index_material(db, mid, row["name"], row["extracted_text"] or "", row["summary"] or "", kw)
db.commit()
except Exception as e:
db.execute("UPDATE materials SET status='error' WHERE id=?", (mid,))
db.commit()
_tasks[key]["error"] = f"{mat['name']}: {e}"
_tasks[key]["done"] = i
_tasks[key]["msg"] = f"已分析 {i}/{len(mats)}"
except Exception as e:
_tasks[key]["error"] = str(e)
finally:
_tasks[key]["running"] = False
if not _tasks[key].get("error"):
_tasks[key]["msg"] = "完成"
db.close()
def _run_project_summary(project_id):
key = _task_key("summary", project_id)
_tasks[key] = {"running": True, "done": 0, "total": 1, "msg": "生成中", "error": None}
db = sqlite3.connect(str(DB_PATH))
db.row_factory = sqlite3.Row
try:
proj = db.execute("SELECT * FROM projects WHERE id=?", (project_id,)).fetchone()
mats = db.execute("SELECT * FROM materials WHERE project_id=?", (project_id,)).fetchall()
# 汇总：已有素材级分析的摘要 + 原始文本
parts = []
for m in mats[:15]:
ana = db.execute("SELECT * FROM analyses WHERE material_id=? AND scope='material' ORDER BY id DESC LIMIT 1", (m["id"],)).fetchone()
if ana and ana["summary"]:
parts.append(f"- {m['name']}: {ana['summary']}")
else:
t = (m["extracted_text"] or "")[:300]
parts.append(f"- {m['name']}: {t}")
content = "\n".join(parts) or "（项目暂无素材）"
prompt = f"""你是项目管理助理。基于以下素材库项目的资料汇总，生成项目级分析报告。只输出合法 JSON：
{{"summary":"整个项目内容的总摘要(150字以内)","key_points":["项目关键发现"...最多8条],"keywords":["项目级关键词"...最多10个],"tags":["项目标签"...最多5个],"category":"建议项目类别"}}
项目名称：{proj['name']}
项目描述：{proj['description'] or '无'}
素材数量：{len(mats)}
资料汇总：
{content[:MAX_TEXT_LEN]}
"""
out = _llm_chat([{"role": "user", "content": prompt}])
data = _parse_json(out)
_store_analysis(db, None, project_id, "project", data, LLM_MODEL, {"model": LLM_MODEL, "raw": out[:3000]})
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), project_id))
db.commit()
_tasks[key]["done"] = 1
_tasks[key]["msg"] = "完成"
except Exception as e:
_tasks[key]["error"] = str(e)
finally:
_tasks[key]["running"] = False
db.close()
# ---------------- 辅助 ----------------
def project_json(row, db):
n = db.execute("SELECT COUNT(*) FROM materials WHERE project_id=?", (row["id"],)).fetchone()[0]
na = db.execute("SELECT COUNT(*) FROM materials WHERE project_id=? AND status='analyzed'", (row["id"],)).fetchone()[0]
ana = db.execute("SELECT * FROM analyses WHERE project_id=? AND scope='project' ORDER BY id DESC LIMIT 1", (row["id"],)).fetchone()
ana_d = None
if ana:
ana_d = dict(ana)
ana_d["key_points"] = json.loads(ana_d["key_points"] or "[]")
ana_d["keywords"] = json.loads(ana_d["keywords"] or "[]")
ana_d["tags"] = json.loads(ana_d["tags"] or "[]")
ana_d["detail"] = json.loads(ana_d["detail"] or "{}")
return {
"id": row["id"], "name": row["name"], "description": row["description"],
"category": row["category"], "tags": row["tags"], "created_at": row["created_at"],
"updated_at": row["updated_at"], "material_count": n, "analyzed_count": na,
"analysis": ana_d,
}
def material_json(row, db):
d = dict(row)
d["meta"] = json.loads(row["meta_json"] or "{}")
ana = db.execute("SELECT * FROM analyses WHERE material_id=? AND scope='material' ORDER BY id DESC LIMIT 1", (row["id"],)).fetchone()
d["analysis"] = dict(ana) if ana else None
if d["analysis"]:
d["analysis"]["key_points"] = json.loads(d["analysis"]["key_points"] or "[]")
d["analysis"]["keywords"] = json.loads(d["analysis"]["keywords"] or "[]")
d["analysis"]["tags"] = json.loads(d["analysis"]["tags"] or "[]")
d["analysis"]["detail"] = json.loads(d["analysis"]["detail"] or "{}")
return d
# ---------------- 路由：页面 ----------------
@app.route("/")
def index():
return send_file(str(STATIC_DIR / "index.html"))
@app.route("/files/<int:mid>")
def serve_file(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row or not row["file_path"]:
abort(404)
path = Path(row["file_path"])
if not path.exists():
abort(404)
ext = (row["ext"] or "").lower()
if ext in IMAGE_EXTS:
return send_file(str(path), mimetype="image/*", download_name=row["name"])
if ext in VIDEO_EXTS:
return send_file(str(path), mimetype="video/mp4", download_name=row["name"])
if ext in AUDIO_EXTS:
return send_file(str(path), mimetype="audio/mpeg", download_name=row["name"])
return send_file(str(path), as_attachment=True, download_name=row["original_name"] or row["name"])
# ---------------- 路由：项目 ----------------
@app.get("/api/projects")
def list_projects():
db = get_db()
q = (request.args.get("q") or "").strip()
category = (request.args.get("category") or "").strip()
tag = (request.args.get("tag") or "").strip()
sql = "SELECT * FROM projects WHERE 1=1"
args = []
if q:
sql += " AND (name LIKE ? OR description LIKE ? OR tags LIKE ?)"
like = f"%{q}%"
args += [like, like, like]
if category:
sql += " AND category=?"
args.append(category)
if tag:
sql += " AND tags LIKE ?"
args.append(f"%{tag}%")
sql += " ORDER BY updated_at DESC, id DESC"
rows = db.execute(sql, args).fetchall()
return jsonify([project_json(r, db) for r in rows])
@app.post("/api/projects")
def create_project():
d = request.get_json(force=True, silent=True) or {}
name = (d.get("name") or "").strip()
if not name:
return jsonify({"error": "项目名称不能为空"}), 400
db = get_db()
cur = db.execute("INSERT INTO projects (name, description, category, tags, updated_at) VALUES (?,?,?,?,?)",
(name, (d.get("description") or "").strip(),
(d.get("category") or "未分类").strip(),
(d.get("tags") or "").strip(), _ts()))
db.commit()
row = db.execute("SELECT * FROM projects WHERE id=?", (cur.lastrowid,)).fetchone()
return jsonify(project_json(row, db))
@app.get("/api/projects/<int:pid>")
def get_project(pid):
db = get_db()
row = db.execute("SELECT * FROM projects WHERE id=?", (pid,)).fetchone()
if not row:
return jsonify({"error": "项目不存在"}), 404
p = project_json(row, db)
mats = db.execute("SELECT * FROM materials WHERE project_id=? ORDER BY id DESC", (pid,)).fetchall()
p["materials"] = [material_json(m, db) for m in mats]
return jsonify(p)
@app.put("/api/projects/<int:pid>")
def update_project(pid):
d = request.get_json(force=True, silent=True) or {}
db = get_db()
row = db.execute("SELECT * FROM projects WHERE id=?", (pid,)).fetchone()
if not row:
return jsonify({"error": "项目不存在"}), 404
name = (d.get("name") or row["name"]).strip()
db.execute("UPDATE projects SET name=?, description=?, category=?, tags=?, updated_at=? WHERE id=?",
(name, (d.get("description", row["description"]) or "").strip(),
(d.get("category", row["category"]) or "未分类").strip(),
(d.get("tags", row["tags"]) or "").strip(), _ts(), pid))
db.commit()
return jsonify(project_json(db.execute("SELECT * FROM projects WHERE id=?", (pid,)).fetchone(), db))
@app.delete("/api/projects/<int:pid>")
def delete_project(pid):
db = get_db()
mats = db.execute("SELECT * FROM materials WHERE project_id=?", (pid,)).fetchall()
for m in mats:
if m["file_path"]:
p = Path(m["file_path"])
if p.exists():
try:
p.unlink()
except Exception:
pass
db.execute("DELETE FROM materials_fts WHERE rowid=?", (m["id"],))
db.execute("DELETE FROM analyses WHERE project_id=?", (pid,))
db.execute("DELETE FROM materials WHERE project_id=?", (pid,))
db.execute("DELETE FROM projects WHERE id=?", (pid,))
db.commit()
return jsonify({"ok": True})
@app.get("/api/meta")
def get_meta():
db = get_db()
cats = db.execute("SELECT category, COUNT(*) c FROM projects GROUP BY category ORDER BY c DESC").fetchall()
all_tags = [r["tags"] for r in db.execute("SELECT tags FROM projects").fetchall() if r["tags"]]
tag_counter = {}
for t in all_tags:
for x in [x.strip() for x in t.split(",") if x.strip()]:
tag_counter[x] = tag_counter.get(x, 0) + 1
stats = {
"projects": db.execute("SELECT COUNT(*) FROM projects").fetchone()[0],
"materials": db.execute("SELECT COUNT(*) FROM materials").fetchone()[0],
"analyzed": db.execute("SELECT COUNT(*) FROM materials WHERE status='analyzed'").fetchone()[0],
}
return jsonify({
"categories": [{"name": r["category"], "count": r["c"]} for r in cats],
"tags": [{"name": k, "count": v} for k, v in sorted(tag_counter.items(), key=lambda x: -x[1])[:40]],
"stats": stats,
})
# ---------------- 路由：素材 ----------------
@app.post("/api/projects/<int:pid>/materials/upload")
def upload_materials(pid):
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
files = request.files.getlist("files")
results = []
for f in files:
if not f or not f.filename:
continue
orig = f.filename
ext = os.path.splitext(orig)[1].lower()
safe = _safe_filename(orig)
if ext not in ALLOWED_EXT:
results.append({"name": orig, "ok": False, "error": f"不支持的文件类型 {ext or '(无扩展名)'}"})
continue
pid_dir = UPLOAD_DIR / str(pid)
pid_dir.mkdir(parents=True, exist_ok=True)
uniq = uuid.uuid4().hex[:10]
dest = pid_dir / f"{uniq}_{safe}"
f.save(str(dest))
mtype = TYPE_BY_EXT.get(ext, "other")
# 提取文本
text, ok = extract_text(dest, ext)
text_status = "extracted" if ok else ("empty" if not text else "error")
# 元数据
meta = {}
if mtype == "image":
meta = _meta_image(dest)
size = dest.stat().st_size
cur = db.execute(
"INSERT INTO materials (project_id, name, mtype, original_name, file_path, file_size, ext, meta_json, extracted_text, text_status) "
"VALUES (?,?,?,?,?,?,?,?,?,?)",
(pid, safe, mtype, orig, str(dest), size, ext, json.dumps(meta, ensure_ascii=False), text, text_status))
mid = cur.lastrowid
_index_material(db, mid, safe, text)
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), pid))
db.commit()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
results.append(material_json(row, db))
return jsonify({"results": results, "count": len(results)})
@app.post("/api/projects/<int:pid>/materials")
def add_text_material(pid):
d = request.get_json(force=True, silent=True) or {}
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
name = (d.get("name") or "").strip() or "未命名文本"
content = (d.get("content") or "")
cur = db.execute(
"INSERT INTO materials (project_id, name, mtype, original_name, file_size, meta_json, extracted_text, text_status, status) "
"VALUES (?,?,?,?,?,?,?,?,?)",
(pid, name, "text", "", len(content.encode("utf-8")), "{}", content,
"extracted" if content.strip() else "empty", "uploaded"))
mid = cur.lastrowid
_index_material(db, mid, name, content)
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), pid))
db.commit()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
return jsonify(material_json(row, db))
@app.get("/api/materials/<int:mid>")
def get_material(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
return jsonify(material_json(row, db))
@app.put("/api/materials/<int:mid>")
def update_material(mid):
d = request.get_json(force=True, silent=True) or {}
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
name = (d.get("name") or row["name"]).strip()
mtype = (d.get("mtype") or row["mtype"]).strip()
db.execute("UPDATE materials SET name=?, mtype=? WHERE id=?", (name, mtype, mid))
# 更新索引
ana = db.execute("SELECT * FROM analyses WHERE material_id=? AND scope='material' ORDER BY id DESC LIMIT 1", (mid,)).fetchone()
kw = ",".join(json.loads(ana["keywords"] or "[]")) if ana else ""
_index_material(db, mid, name, row["extracted_text"] or "", (ana["summary"] if ana else "") or "", kw)
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), row["project_id"]))
db.commit()
row2 = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
return jsonify(material_json(row2, db))
@app.delete("/api/materials/<int:mid>")
def delete_material(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
if row["file_path"]:
p = Path(row["file_path"])
if p.exists():
try:
p.unlink()
except Exception:
pass
db.execute("DELETE FROM materials_fts WHERE rowid=?", (mid,))
db.execute("DELETE FROM analyses WHERE material_id=?", (mid,))
db.execute("DELETE FROM materials WHERE id=?", (mid,))
db.execute("UPDATE projects SET updated_at=? WHERE id=?", (_ts(), row["project_id"]))
db.commit()
return jsonify({"ok": True})
# ---------------- 路由：分析 ----------------
@app.post("/api/materials/<int:mid>/analyze")
def analyze_one(mid):
db = get_db()
row = db.execute("SELECT * FROM materials WHERE id=?", (mid,)).fetchone()
if not row:
return jsonify({"error": "素材不存在"}), 404
t = threading.Thread(target=_run_analysis, args=(row["project_id"], [mid]), daemon=True)
t.start()
return jsonify({"ok": True, "task": _task_key("material", row["project_id"])})
@app.post("/api/projects/<int:pid>/analyze")
def analyze_project(pid):
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
key = _task_key("material", pid)
t = threading.Thread(target=_run_analysis, args=(pid, None, "material"), daemon=True)
t.start()
return jsonify({"ok": True, "task": key})
@app.post("/api/projects/<int:pid>/summary")
def project_summary(pid):
db = get_db()
if not db.execute("SELECT 1 FROM projects WHERE id=?", (pid,)).fetchone():
return jsonify({"error": "项目不存在"}), 404
t = threading.Thread(target=_run_project_summary, args=(pid,), daemon=True)
t.start()
return jsonify({"ok": True, "task": _task_key("summary", pid)})
@app.get("/api/tasks/<path:key>")
def task_status(key):
t = _tasks.get(key)
if not t:
return jsonify({"running": False, "done": 0, "total": 0, "msg": "无任务", "error": None})
return jsonify(t)
# ---------------- 路由：搜索 ----------------
@app.get("/api/search")
def search():
q = (request.args.get("q") or "").strip()
if not q:
return jsonify({"projects": [], "materials": []})
db = get_db()
# 1) 项目 LIKE
projs = db.execute("SELECT * FROM projects WHERE name LIKE ? OR description LIKE ? OR tags LIKE ? ORDER BY updated_at DESC LIMIT 10",
(f"%{q}%", f"%{q}%", f"%{q}%")).fetchall()
# 2) FTS 素材
materials = []
tokens = [t for t in jieba.cut(q) if t.strip()]
if tokens:
match = " AND ".join('"' + t + '"' for t in tokens[:6])
try:
ids = [r[0] for r in db.execute("SELECT rowid FROM materials_fts WHERE materials_fts MATCH ? ORDER BY rank LIMIT 50", (match,)).fetchall()]
if ids:
ph = ",".join("?" for _ in ids)
rows = db.execute(f"SELECT * FROM materials WHERE id IN ({ph})", ids).fetchall()
by_id = {r["id"]: r for r in rows}
for i in ids:
if i in by_id:
materials.append(by_id[i])
except Exception:
materials = []
# 3) LIKE 兜底
if not materials:
materials = db.execute("SELECT * FROM materials WHERE name LIKE ? OR extracted_text LIKE ? ORDER BY id DESC LIMIT 30",
(f"%{q}%", f"%{q}%")).fetchall()
return jsonify({
"projects": [project_json(p, db) for p in projs],
"materials": [material_json(m, db) for m in materials],
})
app.teardown_appcontext(close_db)
if __name__ == "__main__":
init_db()
port = int(os.environ.get("PORT", "16091"))
print(f"素材库系统启动: http://0.0.0.0:{port}")
app.run(host="0.0.0.0", port=port, debug=False, threaded=True)
Reference in New Issue Copy Permalink