105 lines
4.2 KiB
Python
105 lines
4.2 KiB
Python
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")
|