feat: 移除 CSV 和 Markdown 导出

2026-07-03 19:14:11 +08:00
parent 4102ecc4e3
commit 0b186fa9f8
9 changed files with 31 additions and 289 deletions
@@ -1,104 +0,0 @@
-import csv
-import io
-import json
-import re
-
-from sqlalchemy import select
-from sqlalchemy.orm import Session
-
-from app.models import Comment, ContentItem, Hotspot, Report
-
-
-def labels_to_text(labels_json: str | None) -> str:
-    try:
-        labels = json.loads(labels_json or "[]")
-    except json.JSONDecodeError:
-        return ""
-    return "，".join(str(label) for label in labels) if isinstance(labels, list) else ""
-
-
-def safe_csv_field(value) -> str:
-    text = "" if value is None else str(value)
-    text = text.replace("\r\n", " ").replace("\n", " ").replace("\r", " ")
-    if text.startswith(("=", "+", "-", "@")):
-        return f"'{text}"
-    return text
-
-
-def safe_filename(platform: str, task_id: str, hotspot_keyword: str, *, extension: str = "csv") -> str:
-    keyword = hotspot_keyword[:20] + ("..." if len(hotspot_keyword) > 20 else "")
-    name = f"{platform}_{task_id}_{keyword}"
-    name = re.sub(r'[/\\:*?"<>|]+', "_", name)
-    name = re.sub(r"_+", "_", name).strip("_")
-    return f"{name}.{extension}"
-
-
-def export_item_comments_csv(session: Session, item_id: str) -> tuple[bytes, str]:
-    item = session.get(ContentItem, item_id)
-    if item is None:
-        raise ValueError("content item not found")
-    hotspot = session.get(Hotspot, item.hotspot_id)
-    comments = list(session.scalars(select(Comment).where(Comment.content_item_id == item.id)))
-    output = io.StringIO()
-    writer = csv.writer(output)
-    writer.writerow(["平台", "任务 ID", "热点 ID", "热点标题", "内容条目 ID", "内容条目标题", "评论 ID", "评论内容", "情绪倾向", "方向标签", "点赞数", "评论时间"])
-    for comment in comments:
-        writer.writerow(
-            [
-                item.platform,
-                item.task_id,
-                hotspot.id if hotspot else "",
-                hotspot.title if hotspot else "",
-                item.id,
-                item.title or "",
-                comment.source_comment_id or "",
-                safe_csv_field(comment.content),
-                comment.sentiment,
-                labels_to_text(comment.labels),
-                comment.like_count or 0,
-                comment.comment_time or "",
-            ]
-        )
-    filename = safe_filename(item.platform, item.task_id, hotspot.title if hotspot else item.title or "comments")
-    return output.getvalue().encode("utf-8-sig"), filename
-
-
-def export_hotspot_comments_csv(session: Session, hotspot_id: str) -> tuple[bytes, str]:
-    hotspot = session.get(Hotspot, hotspot_id)
-    if hotspot is None:
-        raise ValueError("hotspot not found")
-    comments = list(session.scalars(select(Comment).where(Comment.hotspot_id == hotspot.id)))
-    output = io.StringIO()
-    writer = csv.writer(output)
-    writer.writerow(["平台", "任务 ID", "热点 ID", "热点标题", "内容条目 ID", "内容条目标题", "评论 ID", "评论内容", "情绪倾向", "方向标签", "点赞数", "评论时间"])
-    for comment in comments:
-        item = session.get(ContentItem, comment.content_item_id)
-        writer.writerow(
-            [
-                comment.platform,
-                comment.task_id,
-                hotspot.id,
-                hotspot.title,
-                item.id if item else "",
-                item.title if item else "",
-                comment.source_comment_id or "",
-                safe_csv_field(comment.content),
-                comment.sentiment,
-                labels_to_text(comment.labels),
-                comment.like_count or 0,
-                comment.comment_time or "",
-            ]
-        )
-    return output.getvalue().encode("utf-8-sig"), safe_filename(hotspot.platform, hotspot.task_id, hotspot.title)
-
-
-def export_report_markdown(session: Session, *, report_type: str, hotspot_id: str | None = None, item_id: str | None = None) -> tuple[str, str]:
-    query = select(Report).where(Report.report_type == report_type)
-    if hotspot_id:
-        query = query.where(Report.hotspot_id == hotspot_id)
-    if item_id:
-        query = query.where(Report.content_item_id == item_id)
-    report = session.scalar(query)
-    if report is None:
-        raise ValueError("report not found")
-    return report.markdown_content or report.markdown, safe_filename("report", report.task_id, report.title, extension="md")