import json from datetime import datetime, timezone from sqlalchemy import func, select from sqlalchemy.orm import Session import app.models # noqa: F401 - ensure models are registered from app.core.security import get_password_hash from app.db.session import sync_engine from app.models.audit_log import AuditLog from app.models.comment import Comment from app.models.favorite import Favorite from app.models.point_ledger import PointLedger from app.models.rating import Rating from app.models.report import Report from app.models.spot import Spot, SpotImage from app.models.tag import SpotTag, Tag from app.models.user import User DEMO_PASSWORD = "demo123456" def get_or_create(session: Session, model, filters: dict, defaults: dict | None = None): instance = session.execute(select(model).filter_by(**filters)).scalar_one_or_none() if instance is not None: return instance, False payload = dict(filters) if defaults: payload.update(defaults) instance = model(**payload) session.add(instance) session.flush() return instance, True def ensure_user( session: Session, *, phone: str, email: str, nickname: str, city: str, identity: str, role: str, ): user, _ = get_or_create( session, User, {"phone": phone}, { "email": email, "password_hash": get_password_hash(DEMO_PASSWORD), "nickname": nickname, "city": city, "identity": identity, "role": role, "is_active": True, "avatar_url": f"https://picsum.photos/seed/{phone[-4:]}/300/300", }, ) changed = False expected = { "email": email, "nickname": nickname, "city": city, "identity": identity, "role": role, "is_active": True, } for field, value in expected.items(): if getattr(user, field) != value: setattr(user, field, value) changed = True if not user.password_hash.startswith("$2"): user.password_hash = get_password_hash(DEMO_PASSWORD) changed = True if changed: session.add(user) session.flush() return user def ensure_tag(session: Session, *, name: str, category: str, usage_count: int): tag, _ = get_or_create( session, Tag, {"name": name}, { "category": category, "usage_count": usage_count, "is_active": True, }, ) tag.category = category tag.usage_count = usage_count tag.is_active = True session.add(tag) session.flush() return tag def ensure_spot( session: Session, *, title: str, city: str, longitude: float, latitude: float, description: str, transport: str, best_time: str, difficulty: str, audit_status: str, reject_reason: str | None, creator_id: int, ): spot, _ = get_or_create( session, Spot, {"title": title}, { "city": city, "description": description, "transport": transport, "best_time": best_time, "difficulty": difficulty, "audit_status": audit_status, "reject_reason": reject_reason, "creator_id": creator_id, "location": func.ST_SetSRID(func.ST_MakePoint(longitude, latitude), 4326), }, ) spot.city = city spot.description = description spot.transport = transport spot.best_time = best_time spot.difficulty = difficulty spot.audit_status = audit_status spot.reject_reason = reject_reason spot.creator_id = creator_id spot.location = func.ST_SetSRID(func.ST_MakePoint(longitude, latitude), 4326) session.add(spot) session.flush() return spot def ensure_spot_image( session: Session, *, spot_id: int, image_url: str, is_cover: bool, sort_order: int, audit_status: str = "approved", ): image, _ = get_or_create( session, SpotImage, {"spot_id": spot_id, "image_url": image_url}, { "is_cover": is_cover, "sort_order": sort_order, "audit_status": audit_status, }, ) image.is_cover = is_cover image.sort_order = sort_order image.audit_status = audit_status session.add(image) session.flush() return image def ensure_spot_tag(session: Session, *, spot_id: int, tag_id: int): get_or_create(session, SpotTag, {"spot_id": spot_id, "tag_id": tag_id}) def ensure_favorite(session: Session, *, user_id: int, spot_id: int): get_or_create(session, Favorite, {"user_id": user_id, "spot_id": spot_id}) def ensure_rating( session: Session, *, user_id: int, spot_id: int, score: int, short_comment: str, ): rating, _ = get_or_create( session, Rating, {"user_id": user_id, "spot_id": spot_id}, { "score": score, "short_comment": short_comment, }, ) rating.score = score rating.short_comment = short_comment session.add(rating) session.flush() return rating def ensure_comment( session: Session, *, spot_id: int, user_id: int, content: str, parent_id: int | None = None, audit_status: str = "approved", ): comment, _ = get_or_create( session, Comment, { "spot_id": spot_id, "user_id": user_id, "parent_id": parent_id, "content": content, }, {"audit_status": audit_status}, ) comment.audit_status = audit_status session.add(comment) session.flush() return comment def ensure_report( session: Session, *, reporter_id: int, target_type: str, target_id: int, reason: str, status: str, handler_id: int | None, conclusion: str | None, ): report, _ = get_or_create( session, Report, { "reporter_id": reporter_id, "target_type": target_type, "target_id": target_id, "reason": reason, }, { "status": status, "handler_id": handler_id, "conclusion": conclusion, "resolved_at": datetime.now(timezone.utc) if status != "pending" else None, }, ) report.status = status report.handler_id = handler_id report.conclusion = conclusion report.resolved_at = datetime.now(timezone.utc) if status != "pending" else None session.add(report) session.flush() return report def ensure_point_ledger( session: Session, *, user_id: int, change: int, balance: int, reason: str, ref_type: str | None, ref_id: int | None, ): ledger, _ = get_or_create( session, PointLedger, { "user_id": user_id, "change": change, "balance": balance, "reason": reason, "ref_type": ref_type, "ref_id": ref_id, }, ) session.add(ledger) session.flush() return ledger def ensure_audit_log( session: Session, *, operator_id: int, action: str, target_type: str, target_id: int | None, detail: dict | None, ): detail_text = json.dumps(detail, ensure_ascii=False) if detail is not None else None log, _ = get_or_create( session, AuditLog, { "operator_id": operator_id, "action": action, "target_type": target_type, "target_id": target_id, "detail": detail_text, }, ) session.add(log) session.flush() return log def refresh_spot_rating_stats(session: Session, spot: Spot): rows = session.execute(select(Rating).where(Rating.spot_id == spot.id)).scalars().all() if rows: spot.rating_count = len(rows) spot.avg_rating = round(sum(item.score for item in rows) / len(rows), 2) else: spot.rating_count = 0 spot.avg_rating = None session.add(spot) session.flush() def seed(): with Session(sync_engine) as session: admin = ensure_user( session, phone="13900000001", email="demo.admin@ciyuan.local", nickname="演示管理员", city="上海", identity="both", role="admin", ) moderator = ensure_user( session, phone="13900000002", email="demo.moderator@ciyuan.local", nickname="演示审核员", city="杭州", identity="both", role="moderator", ) coser = ensure_user( session, phone="13900000003", email="demo.coser@ciyuan.local", nickname="演示Coser", city="上海", identity="coser", role="user", ) photographer = ensure_user( session, phone="13900000004", email="demo.photo@ciyuan.local", nickname="演示摄影师", city="苏州", identity="photographer", role="user", ) traveler = ensure_user( session, phone="13900000005", email="demo.travel@ciyuan.local", nickname="演示旅拍用户", city="南京", identity="both", role="user", ) sakura = ensure_tag(session, name="演示-樱花", category="季节", usage_count=12) night = ensure_tag(session, name="演示-夜景", category="氛围", usage_count=8) cyber = ensure_tag(session, name="演示-赛博", category="风格", usage_count=6) hanfu = ensure_tag(session, name="演示-古风", category="风格", usage_count=10) street = ensure_tag(session, name="演示-街拍", category="题材", usage_count=9) spot_1 = ensure_spot( session, title="演示-上海静安樱花天桥", city="上海", longitude=121.4452, latitude=31.2336, description="春季樱花盛开时非常适合轻婚纱、JK 和日系角色外拍，桥面视野开阔。", transport="地铁 2 号线步行 8 分钟可达，附近可打车临停。", best_time="3 月下旬到 4 月上旬，清晨 7:00-9:00", difficulty="人流中等，需要提早到场占机位。", audit_status="approved", reject_reason=None, creator_id=coser.id, ) spot_2 = ensure_spot( session, title="演示-苏州工业园区玻璃连廊", city="苏州", longitude=120.7304, latitude=31.3241, description="现代感很强的玻璃连廊，适合赛博、都市未来风格拍摄。", transport="自驾更方便，园区停车位充足。", best_time="傍晚蓝调时刻到入夜后 1 小时", difficulty="夜间补光需求较高。", audit_status="approved", reject_reason=None, creator_id=photographer.id, ) spot_3 = ensure_spot( session, title="演示-南京城墙古风角楼", city="南京", longitude=118.8037, latitude=32.0647, description="城墙转角的古风点位，适合汉服、武侠、国风角色。", transport="地铁直达，步行约 10 分钟。", best_time="上午逆光柔和或傍晚金色时段", difficulty="游客较多，需要避开节假日。", audit_status="pending", reject_reason=None, creator_id=traveler.id, ) spot_4 = ensure_spot( session, title="演示-杭州废墟工业仓", city="杭州", longitude=120.1551, latitude=30.2741, description="老工业仓改造区域，墙体纹理丰富，适合废土和剧情向拍摄。", transport="建议包车前往，器材搬运更方便。", best_time="阴天全天都适合，层次更好。", difficulty="地面不平整，服装和脚下要注意。", audit_status="rejected", reject_reason="场地方临时封闭，当前不允许外拍。", creator_id=photographer.id, ) ensure_spot_image( session, spot_id=spot_1.id, image_url="https://picsum.photos/seed/demo-spot-1-cover/1200/800", is_cover=True, sort_order=0, ) ensure_spot_image( session, spot_id=spot_1.id, image_url="https://picsum.photos/seed/demo-spot-1-2/1200/800", is_cover=False, sort_order=1, ) ensure_spot_image( session, spot_id=spot_2.id, image_url="https://picsum.photos/seed/demo-spot-2-cover/1200/800", is_cover=True, sort_order=0, ) ensure_spot_image( session, spot_id=spot_3.id, image_url="https://picsum.photos/seed/demo-spot-3-cover/1200/800", is_cover=True, sort_order=0, audit_status="pending", ) ensure_spot_image( session, spot_id=spot_4.id, image_url="https://picsum.photos/seed/demo-spot-4-cover/1200/800", is_cover=True, sort_order=0, audit_status="rejected", ) ensure_spot_tag(session, spot_id=spot_1.id, tag_id=sakura.id) ensure_spot_tag(session, spot_id=spot_1.id, tag_id=street.id) ensure_spot_tag(session, spot_id=spot_2.id, tag_id=night.id) ensure_spot_tag(session, spot_id=spot_2.id, tag_id=cyber.id) ensure_spot_tag(session, spot_id=spot_3.id, tag_id=hanfu.id) ensure_spot_tag(session, spot_id=spot_4.id, tag_id=street.id) ensure_favorite(session, user_id=coser.id, spot_id=spot_2.id) ensure_favorite(session, user_id=photographer.id, spot_id=spot_1.id) ensure_favorite(session, user_id=traveler.id, spot_id=spot_1.id) ensure_rating( session, user_id=photographer.id, spot_id=spot_1.id, score=5, short_comment="樱花季氛围特别好，出片稳定。", ) ensure_rating( session, user_id=traveler.id, spot_id=spot_1.id, score=4, short_comment="人稍微有点多，但构图空间不错。", ) ensure_rating( session, user_id=coser.id, spot_id=spot_2.id, score=5, short_comment="夜景和反光材质非常适合赛博片。", ) refresh_spot_rating_stats(session, spot_1) refresh_spot_rating_stats(session, spot_2) refresh_spot_rating_stats(session, spot_3) refresh_spot_rating_stats(session, spot_4) comment_1 = ensure_comment( session, spot_id=spot_1.id, user_id=traveler.id, content="工作日早上人真的少很多，推荐 7 点前到。", ) ensure_comment( session, spot_id=spot_1.id, user_id=coser.id, parent_id=comment_1.id, content="收到，下次准备带反光板去拍。", ) ensure_comment( session, spot_id=spot_2.id, user_id=photographer.id, content="现场环境偏暗，建议准备外拍灯和脚架。", ) resolved_report = ensure_report( session, reporter_id=traveler.id, target_type="spot", target_id=spot_4.id, reason="演示上报：该地点当前施工封闭，信息需要更新。", status="resolved", handler_id=moderator.id, conclusion="已核实并驳回点位展示，等待重新开放。", ) pending_report = ensure_report( session, reporter_id=coser.id, target_type="comment", target_id=comment_1.id, reason="演示上报：评论里包含错误时间信息。", status="pending", handler_id=None, conclusion=None, ) ensure_point_ledger( session, user_id=coser.id, change=10, balance=10, reason="演示地点审核通过奖励", ref_type="spot_approved", ref_id=spot_1.id, ) ensure_point_ledger( session, user_id=photographer.id, change=6, balance=6, reason="演示评分与评论奖励", ref_type="engagement_bonus", ref_id=spot_2.id, ) ensure_point_ledger( session, user_id=traveler.id, change=3, balance=3, reason="演示有效反馈奖励", ref_type="report_reward", ref_id=spot_4.id, ) ensure_audit_log( session, operator_id=moderator.id, action="spot.approve", target_type="spot", target_id=spot_1.id, detail={"note": "演示数据：点位信息完整，允许展示"}, ) ensure_audit_log( session, operator_id=moderator.id, action="spot.reject", target_type="spot", target_id=spot_4.id, detail={"reason": "演示数据：场地方封闭"}, ) ensure_audit_log( session, operator_id=admin.id, action="report.resolve", target_type="report", target_id=resolved_report.id, detail={"result": "演示工单已关闭"}, ) ensure_audit_log( session, operator_id=admin.id, action="report.review", target_type="report", target_id=pending_report.id, detail={"result": "演示工单待处理"}, ) session.commit() print("Demo data ready.") print(f"Demo password for all demo users: {DEMO_PASSWORD}") for table, model in [ ("users", User), ("spots", Spot), ("spot_images", SpotImage), ("favorites", Favorite), ("point_ledger", PointLedger), ("audit_logs", AuditLog), ("comments", Comment), ("reports", Report), ("ratings", Rating), ("tags", Tag), ("spot_tags", SpotTag), ]: total = session.execute(select(func.count()).select_from(model)).scalar_one() print(f"{table}: {total}") if __name__ == "__main__": seed()