""" Admin panel business logic. Sections: - Shared uploads & DB sync - Community posts - Rewards - Member notifications - Content (Privacy / Terms / FAQ) """ from __future__ import annotations import uuid from datetime import datetime, timezone from zoneinfo import ZoneInfo from pathlib import Path from fastapi import HTTPException, UploadFile, status from sqlalchemy import inspect, or_, text, func from sqlalchemy.engine import Engine from sqlalchemy.orm import Session, joinedload from app.config import get_settings from app.models import User, UserRole from app.models_admin import ( CommunityPost, CommunityPostComment, CommunityPostLike, CommunityPostShare, ContentPage, ContentPageSlug, Faq, FaqStatus, MemberPoints, MemberSessionProgress, MemberSessionProgressImage, MemberNotification, MemberNotificationType, Reward, RewardClaim, ) from app.schemas import MemberSessionProgressSaveRequest, RewardClaimRequest def get_member_points(db: Session, user_id: int) -> dict | None: """Return the member_points row for a user if present. Returns a dict with keys: user_id, assigned_points_total, available_points or None if no row exists. """ try: row = db.execute(text("SELECT user_id, assigned_points_total, available_points FROM member_points WHERE user_id = :uid"), {"uid": user_id}).fetchone() except Exception: db.rollback() return None if not row: return None return {"user_id": int(row.user_id), "assigned_points_total": int(row.assigned_points_total), "available_points": int(row.available_points)} def get_member_points_bulk(db: Session, user_ids: list[int]) -> dict[int, dict]: """Bulk read cached points from member_points table (fast path).""" ids = [int(uid) for uid in (user_ids or []) if int(uid) > 0] if not ids: return {} try: rows = db.execute( text( "SELECT user_id, assigned_points_total, available_points " "FROM member_points WHERE user_id = ANY(:uids)" ), {"uids": ids}, ).fetchall() except Exception: db.rollback() return {} out: dict[int, dict] = {} for r in rows: uid = int(r.user_id) out[uid] = { "user_id": uid, "assigned_points_total": int(r.assigned_points_total), "available_points": int(r.available_points), } return out def calculate_member_points(db: Session, user_id: int) -> dict: assigned_sum = db.query(func.coalesce(func.sum(MemberSessionProgress.assigned_points), 0)).filter( MemberSessionProgress.user_id == user_id ).scalar() or 0 claimed_sum = db.query(func.coalesce(func.sum(RewardClaim.total_points), 0)).filter( RewardClaim.user_id == user_id, _reward_claim_reserves_points(RewardClaim.status), ).scalar() or 0 assigned_points_total = int(assigned_sum) available_points = assigned_points_total - int(claimed_sum) return { "user_id": user_id, "assigned_points_total": assigned_points_total, "available_points": available_points, } def calculate_member_points_bulk(db: Session, user_ids: list[int]) -> dict[int, dict]: """Compute points for many users in 2 aggregate queries (no N+1).""" ids = [int(uid) for uid in (user_ids or []) if int(uid) > 0] if not ids: return {} # Fast path: use cached member_points when available. cached = get_member_points_bulk(db, ids) missing = [uid for uid in ids if uid not in cached] if not missing: return cached assigned_rows = ( db.query(MemberSessionProgress.user_id, func.coalesce(func.sum(MemberSessionProgress.assigned_points), 0)) .filter(MemberSessionProgress.user_id.in_(missing)) .group_by(MemberSessionProgress.user_id) .all() ) assigned_by_user = {int(uid): int(total or 0) for uid, total in assigned_rows} claimed_rows = ( db.query(RewardClaim.user_id, func.coalesce(func.sum(RewardClaim.total_points), 0)) .filter( RewardClaim.user_id.in_(missing), _reward_claim_reserves_points(RewardClaim.status), ) .group_by(RewardClaim.user_id) .all() ) claimed_by_user = {int(uid): int(total or 0) for uid, total in claimed_rows} out: dict[int, dict] = {**cached} for uid in missing: assigned_points_total = int(assigned_by_user.get(uid, 0)) available_points = assigned_points_total - int(claimed_by_user.get(uid, 0)) out[uid] = { "user_id": uid, "assigned_points_total": assigned_points_total, "available_points": available_points, } return out def refresh_member_points(db: Session, user_id: int) -> MemberPoints: """Recalculate and cache a member's assigned and available points.""" points = calculate_member_points(db, user_id) row = db.query(MemberPoints).filter(MemberPoints.user_id == user_id).first() if row is None: row = MemberPoints(user_id=user_id) db.add(row) row.assigned_points_total = points["assigned_points_total"] row.available_points = points["available_points"] return row # --------------------------------------------------------------------------- # Shared # --------------------------------------------------------------------------- ALLOWED_COMMUNITY_MEDIA_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp", ".mp4", ".mov", ".webm"} ALLOWED_COMMUNITY_MEDIA_CONTENT_TYPES = { "image/jpeg", "image/png", "image/gif", "image/webp", "video/mp4", "video/quicktime", "video/webm", } ALLOWED_REWARD_IMAGE_EXTENSIONS = {".jpg", ".jpeg", ".png", ".gif", ".webp"} ALLOWED_REWARD_IMAGE_CONTENT_TYPES = {"image/jpeg", "image/png", "image/gif", "image/webp"} def repo_root() -> Path: return Path(__file__).resolve().parents[2] def uploads_base_dir() -> Path: settings = get_settings() raw = (settings.community_upload_dir or "uploads").strip() base = Path(raw) if not base.is_absolute(): base = repo_root() / base base.mkdir(parents=True, exist_ok=True) return base def media_public_url(relative_path: str | None) -> str | None: if not relative_path: return None settings = get_settings() prefix = settings.http_api_prefix.rstrip("/") uploads_segment = settings.community_uploads_url_segment.strip("/") return f"{prefix}/{uploads_segment}/{relative_path.lstrip('/')}" def sync_admin_schemas(engine: Engine) -> None: """Run all admin table sync helpers on startup.""" sync_community_post_columns(engine) sync_rewards_schema(engine) sync_reward_claims_schema(engine) sync_member_session_progress_schema(engine) sync_member_session_progress_images_schema(engine) sync_member_notifications_schema(engine) from app.services.admin_notification_service import sync_admin_notifications_schema sync_admin_notifications_schema(engine) sync_content_schema(engine) sync_member_points_schema(engine) from app.services.member_inactive_followup_service import ( sync_member_inactive_followups_schema, sync_users_last_active_at_column, ) from app.services.user_profile_image_service import sync_users_profile_image_column sync_users_last_active_at_column(engine) sync_users_profile_image_column(engine) sync_member_inactive_followups_schema(engine) # --------------------------------------------------------------------------- # Community # --------------------------------------------------------------------------- def community_upload_dir() -> Path: dest = uploads_base_dir() / "community" dest.mkdir(parents=True, exist_ok=True) return dest async def save_community_media(media: UploadFile) -> tuple[str, str]: settings = get_settings() content_type = (media.content_type or "").split(";")[0].strip().lower() filename = media.filename or "" ext = Path(filename).suffix.lower() if not ext and content_type.startswith("image/"): ext = "." + content_type.split("/", 1)[1] if ext not in ALLOWED_COMMUNITY_MEDIA_EXTENSIONS: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported media type. Allowed: {', '.join(sorted(ALLOWED_COMMUNITY_MEDIA_EXTENSIONS))}", ) if content_type and content_type not in ALLOWED_COMMUNITY_MEDIA_CONTENT_TYPES: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported media content type") data = await media.read() max_bytes = settings.community_max_upload_mb * 1024 * 1024 if len(data) > max_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Media file too large (max {settings.community_max_upload_mb} MB)", ) if not data: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Empty media file") stored_name = f"{uuid.uuid4().hex}{ext}" dest = community_upload_dir() / stored_name dest.write_bytes(data) return f"community/{stored_name}", content_type or "application/octet-stream" _COMMUNITY_POST_COLUMN_DDLS: tuple[tuple[str, str], ...] = ( ("title", "VARCHAR(255) NOT NULL DEFAULT ''"), ("post_type", "VARCHAR(64) NOT NULL DEFAULT 'general'"), ("content", "TEXT NOT NULL DEFAULT ''"), ("media_path", "VARCHAR(512)"), ("media_content_type", "VARCHAR(128)"), ("created_by_id", "INTEGER"), ("created_at", "TIMESTAMP WITH TIME ZONE DEFAULT NOW()"), ("updated_at", "TIMESTAMP WITH TIME ZONE DEFAULT NOW()"), ) _CREATE_COMMUNITY_POSTS_PG = """ CREATE TABLE IF NOT EXISTS community_posts ( id SERIAL PRIMARY KEY, title VARCHAR(255) NOT NULL DEFAULT '', post_type VARCHAR(64) NOT NULL DEFAULT 'general', content TEXT NOT NULL DEFAULT '', media_path VARCHAR(512), media_content_type VARCHAR(128), created_by_id INTEGER, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); CREATE INDEX IF NOT EXISTS ix_community_posts_id ON community_posts (id); """ def _sync_community_post_columns(engine: Engine) -> None: try: insp = inspect(engine) if "community_posts" not in insp.get_table_names(): return cols = {c["name"] for c in insp.get_columns("community_posts")} except Exception: return is_pg = engine.dialect.name == "postgresql" with engine.begin() as conn: for name, ddl in _COMMUNITY_POST_COLUMN_DDLS: if name in cols: continue if is_pg: conn.execute(text(f"ALTER TABLE community_posts ADD COLUMN IF NOT EXISTS {name} {ddl}")) else: try: conn.execute(text(f"ALTER TABLE community_posts ADD COLUMN {name} {ddl}")) except Exception: pass if not is_pg: return try: insp = inspect(engine) fks = {fk["name"] for fk in insp.get_foreign_keys("community_posts")} except Exception: return if "community_posts_created_by_id_fkey" not in fks: try: with engine.begin() as conn: conn.execute( text( """ ALTER TABLE community_posts ADD CONSTRAINT community_posts_created_by_id_fkey FOREIGN KEY (created_by_id) REFERENCES users(id) ON DELETE SET NULL """ ) ) except Exception: pass _fix_post_type_enum_to_varchar(engine) def _fix_post_type_enum_to_varchar(engine: Engine) -> None: if engine.dialect.name != "postgresql": return try: with engine.connect() as conn: row = conn.execute( text( """ SELECT udt_name FROM information_schema.columns WHERE table_schema = 'public' AND table_name = 'community_posts' AND column_name = 'post_type' """ ) ).fetchone() except Exception: return if not row or row[0] != "communityposttype": return with engine.begin() as conn: conn.execute( text( "ALTER TABLE community_posts ALTER COLUMN post_type TYPE VARCHAR(64) USING post_type::text" ) ) conn.execute(text("ALTER TABLE community_posts ALTER COLUMN post_type SET DEFAULT 'general'")) def sync_community_post_columns(engine: Engine) -> None: is_pg = engine.dialect.name == "postgresql" try: insp = inspect(engine) table_exists = "community_posts" in insp.get_table_names() except Exception: table_exists = False if not table_exists and is_pg: with engine.begin() as conn: conn.execute(text(_CREATE_COMMUNITY_POSTS_PG)) elif not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["community_posts"].create(bind=engine, checkfirst=True) _sync_community_post_columns(engine) from app.database import Base Base.metadata.tables["community_post_likes"].create(bind=engine, checkfirst=True) Base.metadata.tables["community_post_comments"].create(bind=engine, checkfirst=True) Base.metadata.tables["community_post_shares"].create(bind=engine, checkfirst=True) def serialize_community_comment(comment: CommunityPostComment) -> dict: return { "id": comment.id, "post_id": comment.post_id, "user_id": comment.user_id, "user_name": comment.user.full_name if comment.user else None, "comment": comment.comment, "created_at": comment.created_at, "updated_at": comment.updated_at, } def serialize_community_post(post: CommunityPost) -> dict: return { "id": post.id, "title": post.title or "", "post_type": post.post_type, "content": post.content, "media_url": media_public_url(post.media_path), "media_content_type": post.media_content_type, "has_media": bool(post.media_path), "created_by_id": post.created_by_id, "created_by_name": post.created_by.full_name if post.created_by else None, "created_at": post.created_at, "updated_at": post.updated_at, "likes_count": 0, "comments_count": 0, "shares_count": 0, "share_counts": {}, "liked_by_me": False, "comments": [], } def list_community_posts(db) -> list[CommunityPost]: return ( db.query(CommunityPost) .options(joinedload(CommunityPost.created_by)) .order_by(CommunityPost.created_at.desc()) .all() ) def delete_community_media_file(relative_path: str | None) -> None: if not relative_path: return path = community_upload_dir() / Path(relative_path).name if path.is_file(): path.unlink() def create_community_post( *, db, admin: User, title: str, post_type: str, content: str, media_path: str | None = None, media_content_type: str | None = None, ) -> CommunityPost: post = CommunityPost( title=title.strip(), post_type=post_type.strip(), content=content.strip(), media_path=media_path, media_content_type=media_content_type, created_by_id=admin.id, ) db.add(post) db.commit() db.refresh(post) post.created_by = admin return post def update_community_post( *, db, post: CommunityPost, title: str, post_type: str, content: str, media_path: str | None = None, media_content_type: str | None = None, replace_media: bool = False, clear_media: bool = False, ) -> CommunityPost: if replace_media or clear_media: delete_community_media_file(post.media_path) post.title = title.strip() post.post_type = post_type.strip() post.content = content.strip() if replace_media: post.media_path = media_path post.media_content_type = media_content_type elif clear_media: post.media_path = None post.media_content_type = None db.commit() db.refresh(post) return post def delete_community_post(db: Session, post_id: int) -> bool: post = db.get(CommunityPost, post_id) if not post: return False delete_community_media_file(post.media_path) db.delete(post) db.commit() return True def delete_community_comment(db: Session, comment_id: int) -> bool: """Admin-only: permanently delete one community_post_comments row.""" row = db.get(CommunityPostComment, comment_id) if not row: return False db.delete(row) db.commit() return True # --------------------------------------------------------------------------- # Rewards # --------------------------------------------------------------------------- def reward_upload_dir() -> Path: dest = uploads_base_dir() / "rewards" dest.mkdir(parents=True, exist_ok=True) return dest async def save_reward_media(media: UploadFile) -> tuple[str, str]: content_type = (media.content_type or "").split(";")[0].strip().lower() filename = media.filename or "" ext = Path(filename).suffix.lower() if not ext and content_type.startswith("image/"): ext = "." + content_type.split("/", 1)[1] if ext not in ALLOWED_REWARD_IMAGE_EXTENSIONS: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported image type. Allowed: {', '.join(sorted(ALLOWED_REWARD_IMAGE_EXTENSIONS))}", ) if content_type and content_type not in ALLOWED_REWARD_IMAGE_CONTENT_TYPES: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported image content type") settings = get_settings() data = await media.read() max_bytes = settings.community_max_upload_mb * 1024 * 1024 if len(data) > max_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Image too large (max {settings.community_max_upload_mb} MB)", ) if not data: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Empty image file") stored_name = f"{uuid.uuid4().hex}{ext}" dest = reward_upload_dir() / stored_name dest.write_bytes(data) return f"rewards/{stored_name}", content_type or "application/octet-stream" _CREATE_REWARDS_PG = """ CREATE TABLE IF NOT EXISTS rewards ( id SERIAL PRIMARY KEY, name VARCHAR(255) NOT NULL, description TEXT NOT NULL DEFAULT '', points_required INTEGER NOT NULL, category VARCHAR(64) NOT NULL, media_path VARCHAR(512), media_content_type VARCHAR(128), created_by_id INTEGER, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); CREATE INDEX IF NOT EXISTS ix_rewards_id ON rewards (id); """ def sync_rewards_schema(engine: Engine) -> None: is_pg = engine.dialect.name == "postgresql" try: insp = inspect(engine) table_exists = "rewards" in insp.get_table_names() except Exception: table_exists = False if not table_exists and is_pg: with engine.begin() as conn: conn.execute(text(_CREATE_REWARDS_PG)) elif not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["rewards"].create(bind=engine, checkfirst=True) if not is_pg or not table_exists: return try: insp = inspect(engine) fks = {fk["name"] for fk in insp.get_foreign_keys("rewards")} except Exception: return if "rewards_created_by_id_fkey" not in fks: try: with engine.begin() as conn: conn.execute( text( """ ALTER TABLE rewards ADD CONSTRAINT rewards_created_by_id_fkey FOREIGN KEY (created_by_id) REFERENCES users(id) ON DELETE SET NULL """ ) ) except Exception: pass def serialize_reward(reward: Reward) -> dict: return { "id": reward.id, "name": reward.name, "description": reward.description, "points_required": reward.points_required, "category": reward.category, "media_url": media_public_url(reward.media_path), "media_content_type": reward.media_content_type, "has_media": bool(reward.media_path), "created_by_id": reward.created_by_id, "created_by_name": reward.created_by.full_name if reward.created_by else None, "created_at": reward.created_at, "updated_at": reward.updated_at, } def serialize_reward_claim(claim: RewardClaim) -> dict: sent_to = (claim.location or "").strip() or "Admin" user_phone = (claim.user.phone_number or "").strip() if claim.user else "" claim_phone = (claim.phone_number or "").strip() phone_number = user_phone or claim_phone or None return { "id": claim.id, "reward_id": claim.reward_id, "product_name": claim.product_name, "user_id": claim.user_id, "user_name": claim.user_name, "email": claim.email, "phone_number": phone_number, "location": claim.location, "sent_to": sent_to, "total_points": claim.total_points, "status": claim.status, "product_image_url": media_public_url(claim.reward.media_path) if claim.reward else None, "remaining_points": getattr(claim, "remaining_points", None), "assigned_points_total": getattr(claim, "assigned_points_total", None), "available_points": getattr(claim, "available_points", None), "created_at": claim.created_at, "updated_at": claim.updated_at, } def attach_claim_points(db: Session, claim: RewardClaim) -> RewardClaim: points = calculate_member_points(db, claim.user_id) claim.assigned_points_total = points["assigned_points_total"] claim.available_points = points["available_points"] claim.remaining_points = points["available_points"] return claim def delete_reward_media_file(relative_path: str | None) -> None: if not relative_path: return path = reward_upload_dir() / Path(relative_path).name if path.is_file(): path.unlink() def list_rewards(db) -> list[Reward]: return ( db.query(Reward) .options(joinedload(Reward.created_by)) .order_by(Reward.created_at.desc()) .all() ) def create_reward( *, db, admin: User, name: str, description: str, points_required: int, category: str, media_path: str | None = None, media_content_type: str | None = None, ) -> Reward: reward = Reward( name=name.strip(), description=description.strip(), points_required=points_required, category=category.strip(), media_path=media_path, media_content_type=media_content_type, created_by_id=admin.id, ) db.add(reward) db.commit() db.refresh(reward) reward.created_by = admin return reward def update_reward( *, db, reward: Reward, name: str, description: str, points_required: int, category: str, media_path: str | None = None, media_content_type: str | None = None, replace_media: bool = False, clear_media: bool = False, ) -> Reward: if replace_media or clear_media: delete_reward_media_file(reward.media_path) reward.name = name.strip() reward.description = description.strip() reward.points_required = points_required reward.category = category.strip() if replace_media: reward.media_path = media_path reward.media_content_type = media_content_type elif clear_media: reward.media_path = None reward.media_content_type = None db.commit() db.refresh(reward) return reward def delete_reward(db: Session, reward_id: int) -> bool: reward = db.get(Reward, reward_id) if not reward: return False delete_reward_media_file(reward.media_path) db.delete(reward) db.commit() return True def sync_reward_claims_schema(engine: Engine) -> None: try: insp = inspect(engine) table_exists = "reward_claims" in insp.get_table_names() except Exception: table_exists = False if not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["reward_claims"].create(bind=engine, checkfirst=True) return try: cols = {c["name"] for c in inspect(engine).get_columns("reward_claims")} except Exception: return # Add missing columns: location (varchar) and remaining_points (integer) to_add = [] if 'location' not in cols: to_add.append(('location', "VARCHAR(255)")) if 'remaining_points' not in cols: to_add.append(('remaining_points', "INTEGER")) if not to_add: return with engine.begin() as conn: for col_name, ddl in to_add: if engine.dialect.name == 'postgresql': conn.execute(text(f"ALTER TABLE reward_claims ADD COLUMN IF NOT EXISTS {col_name} {ddl}")) else: try: conn.execute(text(f"ALTER TABLE reward_claims ADD COLUMN {col_name} {ddl}")) except Exception: pass def create_reward_claim(db: Session, user: User, body: RewardClaimRequest) -> RewardClaim: reward = db.get(Reward, body.reward_id) if not reward: raise ValueError("Reward not found") existing_claim = ( db.query(RewardClaim) .filter( RewardClaim.user_id == user.id, RewardClaim.reward_id == reward.id, _reward_claim_reserves_points(RewardClaim.status), ) .first() ) if existing_claim: raise ValueError("You have already claimed this reward.") # Compute user's assigned points (earned) from session progress assigned_sum = db.query(func.coalesce(func.sum(MemberSessionProgress.assigned_points), 0)).filter( MemberSessionProgress.user_id == user.id ).scalar() or 0 # Sum points already reserved/claimed by the user (pending or accepted) claimed_sum = db.query(func.coalesce(func.sum(RewardClaim.total_points), 0)).filter( RewardClaim.user_id == user.id, _reward_claim_reserves_points(RewardClaim.status), ).scalar() or 0 available = int(assigned_sum) - int(claimed_sum) requested = int(body.total_points) if body.total_points is not None else int(reward.points_required) if requested > available: # raise ValueError to be handled by router and presented to client raise ValueError(f"Insufficient points. Available: {available}, requested: {requested}") phone_number = (body.phone_number or user.phone_number or "").strip() or None claim = RewardClaim( reward_id=reward.id, user_id=user.id, product_name=reward.name, user_name=user.full_name or "", email=user.email, phone_number=phone_number, location=(body.location or "").strip() or None, total_points=requested, remaining_points=int(available - requested), status="pending", ) db.add(claim) db.flush() points = refresh_member_points(db, user.id) db.commit() db.refresh(claim) claim.reward = reward claim.user = user claim.assigned_points_total = points.assigned_points_total claim.available_points = points.available_points # compute remaining and attach for serialization remaining = available - requested try: setattr(claim, 'remaining_points', int(remaining)) except Exception: pass from app.services.admin_notification_service import notify_admin_reward_claim notify_admin_reward_claim( db, user, product_name=claim.product_name, points=int(claim.total_points), claim_id=claim.id, ) return claim def list_reward_claims(db: Session, *, status_filter: str | None = None) -> list[RewardClaim]: query = db.query(RewardClaim).options( joinedload(RewardClaim.reward), joinedload(RewardClaim.user), ) if status_filter: query = query.filter(RewardClaim.status == status_filter) claims = query.order_by(RewardClaim.created_at.desc(), RewardClaim.id.desc()).all() for claim in claims: attach_claim_points(db, claim) return claims def list_user_reward_claims( db: Session, user_id: int, *, status_filter: str | None = None, ) -> list[RewardClaim]: """Reward claims for one member (user My Claims page).""" query = ( db.query(RewardClaim) .options(joinedload(RewardClaim.reward), joinedload(RewardClaim.user)) .filter(RewardClaim.user_id == user_id) ) if status_filter: query = query.filter(RewardClaim.status == status_filter) return query.order_by(RewardClaim.created_at.desc(), RewardClaim.id.desc()).all() def _normalize_reward_claim_status(status_value: str) -> str: aliases = { "accept": "accepted", "accepted": "accepted", "approve": "accepted", "approved": "accepted", "reject": "rejected", "rejected": "rejected", "pending": "pending", } normalized = (status_value or "").strip().lower() return aliases.get(normalized, normalized) def _reward_claim_reserves_points(status_column): return func.lower(func.trim(status_column)).in_(("pending", "accepted")) def update_reward_claim_status( db: Session, claim_id: int, status_value: str ) -> RewardClaim | None: claim = ( db.query(RewardClaim) .options(joinedload(RewardClaim.reward), joinedload(RewardClaim.user)) .filter(RewardClaim.id == claim_id) .first() ) if not claim: return None normalized_status = _normalize_reward_claim_status(status_value) if normalized_status not in {"pending", "accepted", "rejected"}: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="status must be one of: pending, accepted, rejected", ) claim.status = normalized_status db.flush() points = refresh_member_points(db, claim.user_id) db.flush() claim.remaining_points = points.available_points db.commit() db.refresh(claim) claim.assigned_points_total = points.assigned_points_total claim.available_points = points.available_points claim.remaining_points = points.available_points return claim def delete_reward_claim(db: Session, claim_id: int) -> bool: claim = db.get(RewardClaim, claim_id) if not claim: return False user_id = claim.user_id db.delete(claim) db.flush() refresh_member_points(db, user_id) db.commit() return True def sync_member_session_progress_schema(engine: Engine) -> None: try: insp = inspect(engine) table_exists = "member_session_progress" in insp.get_table_names() except Exception: table_exists = False if not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["member_session_progress"].create(bind=engine, checkfirst=True) return try: cols = {c["name"] for c in inspect(engine).get_columns("member_session_progress")} except Exception: return with engine.begin() as conn: if "image_path" not in cols: if engine.dialect.name == "postgresql": conn.execute( text( "ALTER TABLE member_session_progress " "ADD COLUMN IF NOT EXISTS image_path VARCHAR(512)" ) ) else: try: conn.execute(text("ALTER TABLE member_session_progress ADD COLUMN image_path VARCHAR(512)")) except Exception: pass if "feedback_notified" not in cols: if engine.dialect.name == "postgresql": conn.execute( text( "ALTER TABLE member_session_progress " "ADD COLUMN IF NOT EXISTS feedback_notified BOOLEAN NOT NULL DEFAULT FALSE" ) ) else: try: conn.execute( text( "ALTER TABLE member_session_progress " "ADD COLUMN feedback_notified BOOLEAN NOT NULL DEFAULT 0" ) ) except Exception: pass _CREATE_MEMBER_SESSION_PROGRESS_IMAGES_PG = """ CREATE TABLE IF NOT EXISTS member_session_progress_images ( id SERIAL PRIMARY KEY, progress_id INTEGER NOT NULL REFERENCES member_session_progress(id) ON DELETE CASCADE, image_path VARCHAR(512) NOT NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); CREATE INDEX IF NOT EXISTS ix_member_session_progress_images_progress_id ON member_session_progress_images (progress_id); CREATE INDEX IF NOT EXISTS ix_member_session_progress_images_id ON member_session_progress_images (id); """ def sync_member_session_progress_images_schema(engine: Engine) -> None: is_pg = engine.dialect.name == "postgresql" try: insp = inspect(engine) table_exists = "member_session_progress_images" in insp.get_table_names() except Exception: table_exists = False if not table_exists and is_pg: with engine.begin() as conn: conn.execute(text(_CREATE_MEMBER_SESSION_PROGRESS_IMAGES_PG)) return if not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["member_session_progress_images"].create(bind=engine, checkfirst=True) return def member_progress_upload_dir() -> Path: dest = uploads_base_dir() / "member-progress" dest.mkdir(parents=True, exist_ok=True) return dest async def save_member_progress_image(image: UploadFile) -> str: content_type = (image.content_type or "").split(";", 1)[0].strip().lower() filename = image.filename or "" ext = Path(filename).suffix.lower() if not ext and content_type.startswith("image/"): ext = "." + content_type.split("/", 1)[1] if ext not in ALLOWED_REWARD_IMAGE_EXTENSIONS: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Unsupported image type. Allowed: {', '.join(sorted(ALLOWED_REWARD_IMAGE_EXTENSIONS))}", ) if content_type and content_type not in ALLOWED_REWARD_IMAGE_CONTENT_TYPES: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Unsupported image content type") settings = get_settings() data = await image.read() max_bytes = settings.community_max_upload_mb * 1024 * 1024 if len(data) > max_bytes: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Image too large (max {settings.community_max_upload_mb} MB)", ) if not data: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Empty image file") stored_name = f"{uuid.uuid4().hex}{ext}" dest = member_progress_upload_dir() / stored_name dest.write_bytes(data) return f"member-progress/{stored_name}" def serialize_member_session_progress(row: MemberSessionProgress) -> dict: image_urls: list[str] = [] extra_rows = getattr(row, "images", None) or [] image_urls = [media_public_url(r.image_path) for r in extra_rows if getattr(r, "image_path", None)] if row.image_path: primary_url = media_public_url(row.image_path) if primary_url and primary_url not in image_urls: image_urls.insert(0, primary_url) return { "id": row.id, "user_id": row.user_id, "week_id": row.week_id, "week_name": row.week_name, "week_index": row.week_index, "session_id": row.session_id, "session_name": row.session_name, "session_number": row.session_number, "assigned_points": row.assigned_points, "feedback": row.feedback, "image_url": media_public_url(row.image_path), "image_urls": image_urls, "is_completed": row.is_completed, "assigned_by_id": row.assigned_by_id, "assigned_by_name": row.assigned_by.full_name if row.assigned_by else None, "created_at": row.created_at, "updated_at": row.updated_at, } def add_member_session_progress_images(db: Session, *, progress_id: int, image_paths: list[str]) -> int: cleaned = [p.strip() for p in (image_paths or []) if (p or "").strip()] if not cleaned: return 0 rows = [MemberSessionProgressImage(progress_id=progress_id, image_path=p) for p in cleaned] db.add_all(rows) db.commit() return len(rows) def _clean_optional_text(value: str | None) -> str | None: return (value or "").strip() or None def save_member_session_progress( db: Session, admin: User, body: MemberSessionProgressSaveRequest, image_path: str | None = None, ) -> MemberSessionProgress: member = db.get(User, body.user_id) if not member or member.role != UserRole.user: raise ValueError("Member not found") query = db.query(MemberSessionProgress).filter(MemberSessionProgress.user_id == body.user_id) if body.session_id is not None: query = query.filter(MemberSessionProgress.session_id == body.session_id) elif body.session_number is not None and body.week_index is not None: query = query.filter( MemberSessionProgress.week_index == body.week_index, MemberSessionProgress.session_number == body.session_number, ) else: query = query.filter( MemberSessionProgress.week_name == _clean_optional_text(body.week_name), MemberSessionProgress.session_name == _clean_optional_text(body.session_name), ) row = query.first() if row is None: row = MemberSessionProgress(user_id=body.user_id) db.add(row) row.week_id = body.week_id row.week_name = _clean_optional_text(body.week_name) row.week_index = body.week_index row.session_id = body.session_id row.session_name = _clean_optional_text(body.session_name) row.session_number = body.session_number if body.add_to_existing: existing_points = int(row.assigned_points or 0) row.assigned_points = existing_points + int(body.assigned_points or 0) else: row.assigned_points = body.assigned_points row.feedback = _clean_optional_text(body.feedback) row.is_completed = body.is_completed row.assigned_by_id = admin.id db.flush() if image_path is not None: # Keep backwards-compatible single image pointer and also store in images table. if not row.image_path: row.image_path = image_path db.add(MemberSessionProgressImage(progress_id=row.id, image_path=image_path)) db.flush() refresh_member_points(db, body.user_id) db.commit() db.refresh(row) row.assigned_by = admin return row def list_member_session_progress( db: Session, user_id: int, *, feedback_only: bool = False, ) -> list[MemberSessionProgress]: from app.member_feedback import is_coach_feedback_text query = ( db.query(MemberSessionProgress) .options(joinedload(MemberSessionProgress.assigned_by), joinedload(MemberSessionProgress.images)) .filter(MemberSessionProgress.user_id == user_id) ) rows = query.order_by( MemberSessionProgress.week_index.asc().nullslast(), MemberSessionProgress.session_number.asc().nullslast(), MemberSessionProgress.created_at.asc(), ).all() if feedback_only: return [row for row in rows if is_coach_feedback_text(row.feedback)] return rows # --------------------------------------------------------------------------- # Member notifications # --------------------------------------------------------------------------- _MEMBER_NOTIFICATION_TYPE_LABELS: dict[MemberNotificationType, str] = { MemberNotificationType.announcement: "Announcement", MemberNotificationType.reward: "Reward", MemberNotificationType.feedback: "Feedback", MemberNotificationType.system: "System", MemberNotificationType.achievement: "Achievement", } _CREATE_MEMBER_NOTIFICATIONS_PG = """ CREATE TABLE IF NOT EXISTS member_notifications ( id SERIAL PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, notification_type VARCHAR(64) NOT NULL, title VARCHAR(255) NOT NULL, message TEXT NOT NULL, is_read BOOLEAN NOT NULL DEFAULT FALSE, admin_cleared BOOLEAN NOT NULL DEFAULT FALSE, sent_by_id INTEGER REFERENCES users(id) ON DELETE SET NULL, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); CREATE INDEX IF NOT EXISTS ix_member_notifications_user_id ON member_notifications (user_id); CREATE INDEX IF NOT EXISTS ix_member_notifications_notification_type ON member_notifications (notification_type); CREATE INDEX IF NOT EXISTS ix_member_notifications_id ON member_notifications (id); """ def sync_member_notifications_schema(engine: Engine) -> None: is_pg = engine.dialect.name == "postgresql" try: insp = inspect(engine) table_exists = "member_notifications" in insp.get_table_names() except Exception: table_exists = False if not table_exists and is_pg: with engine.begin() as conn: conn.execute(text(_CREATE_MEMBER_NOTIFICATIONS_PG)) elif not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["member_notifications"].create(bind=engine, checkfirst=True) if not table_exists: return try: cols = {c["name"] for c in inspect(engine).get_columns("member_notifications")} except Exception: return if "admin_cleared" in cols: return with engine.begin() as conn: if is_pg: conn.execute( text( "ALTER TABLE member_notifications " "ADD COLUMN IF NOT EXISTS admin_cleared BOOLEAN NOT NULL DEFAULT FALSE" ) ) else: try: conn.execute( text( "ALTER TABLE member_notifications " "ADD COLUMN admin_cleared BOOLEAN NOT NULL DEFAULT 0" ) ) except Exception: pass _CREATE_MEMBER_POINTS_PG = """ CREATE TABLE IF NOT EXISTS member_points ( id SERIAL PRIMARY KEY, user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE, assigned_points_total INTEGER NOT NULL DEFAULT 0, available_points INTEGER NOT NULL DEFAULT 0, updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); CREATE INDEX IF NOT EXISTS ix_member_points_user_id ON member_points (user_id); """ def sync_member_points_schema(engine: Engine) -> None: """Create or alter the `member_points` table used to cache per-user points.""" is_pg = engine.dialect.name == "postgresql" try: insp = inspect(engine) table_exists = "member_points" in insp.get_table_names() except Exception: table_exists = False if not table_exists and is_pg: with engine.begin() as conn: conn.execute(text(_CREATE_MEMBER_POINTS_PG)) elif not table_exists: from app import models_admin # noqa: F401 from app.database import Base Base.metadata.tables["member_points"].create(bind=engine, checkfirst=True) return try: cols = {c["name"] for c in inspect(engine).get_columns("member_points")} except Exception: return to_add = [] if "assigned_points_total" not in cols: to_add.append(("assigned_points_total", "INTEGER NOT NULL DEFAULT 0")) if "available_points" not in cols: to_add.append(("available_points", "INTEGER NOT NULL DEFAULT 0")) if "updated_at" not in cols: if is_pg: to_add.append(("updated_at", "TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL")) else: to_add.append(("updated_at", "DATETIME")) if not to_add: return with engine.begin() as conn: for col_name, ddl in to_add: if is_pg: conn.execute(text(f"ALTER TABLE member_points ADD COLUMN IF NOT EXISTS {col_name} {ddl}")) else: try: conn.execute(text(f"ALTER TABLE member_points ADD COLUMN {col_name} {ddl}")) except Exception: pass def _normalize_member_notification_type(raw: str) -> MemberNotificationType: key = raw.strip().lower().replace(" ", "_") try: return MemberNotificationType(key) except ValueError: allowed = ", ".join(t.value for t in MemberNotificationType) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Invalid notification type. Allowed: {allowed}", ) from None def _member_notification_type_label(notification_type: str) -> str: try: return _MEMBER_NOTIFICATION_TYPE_LABELS[MemberNotificationType(notification_type)] except ValueError: return notification_type.replace("_", " ").title() def _format_notification_datetime(dt: datetime | None) -> str | None: if dt is None: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) return dt.astimezone(ZoneInfo("Europe/London")).strftime("%d %b, %Y %I:%M %p").lstrip("0") def serialize_member_notification(row: MemberNotification) -> dict: user_name = row.user.full_name if row.user and row.user.full_name else f"User #{row.user_id}" sent_by_name = None if row.sent_by and row.sent_by.full_name: sent_by_name = row.sent_by.full_name elif row.sent_by_id: sent_by_name = f"Admin #{row.sent_by_id}" return { "id": row.id, "user_id": row.user_id, "user_name": user_name, "recipient": user_name, "notification_type": row.notification_type, "type_label": _member_notification_type_label(row.notification_type), "title": row.title, "message": row.message, "is_read": row.is_read, "admin_cleared": row.admin_cleared, "sent_by_id": row.sent_by_id, "sent_by_name": sent_by_name, "created_at": row.created_at, "created_at_label": _format_notification_datetime(row.created_at), } def _admin_visible_notifications_query(db: Session): return db.query(MemberNotification).filter(MemberNotification.admin_cleared.is_(False)) def list_member_notifications( db: Session, *, include_cleared: bool = False ) -> list[MemberNotification]: query = db.query(MemberNotification) if not include_cleared: query = query.filter(MemberNotification.admin_cleared.is_(False)) return ( query.options( joinedload(MemberNotification.user), joinedload(MemberNotification.sent_by), ) .order_by(MemberNotification.created_at.desc(), MemberNotification.id.desc()) .all() ) def delete_member_notification(db: Session, notification_id: int) -> bool: """Permanently delete one notification row from the admin notifications list.""" row = db.get(MemberNotification, notification_id) if not row: return False db.delete(row) db.commit() return True def mark_member_notification_read(db: Session, notification_id: int) -> MemberNotification | None: row = db.get(MemberNotification, notification_id) if not row or row.admin_cleared: return None row.is_read = True db.commit() db.refresh(row) return row def mark_all_member_notifications_read(db: Session) -> int: count = ( _admin_visible_notifications_query(db) .filter(MemberNotification.is_read.is_(False)) .update({MemberNotification.is_read: True}, synchronize_session=False) ) db.commit() return count def clear_all_member_notifications(db: Session) -> int: count = ( _admin_visible_notifications_query(db) .update({MemberNotification.admin_cleared: True}, synchronize_session=False) ) db.commit() return count def count_unread_admin_member_notifications(db: Session) -> int: """Unread count for admin bell badge (visible rows only: admin_cleared=false, is_read=false).""" return ( _admin_visible_notifications_query(db) .filter(MemberNotification.is_read.is_(False)) .count() ) def list_notification_recipients(db: Session) -> list[dict]: members = ( db.query(User) .filter(User.role == UserRole.user, User.is_active.is_(True)) .order_by(User.full_name.asc(), User.id.asc()) .all() ) return [ { "id": m.id, "full_name": m.full_name.strip() or m.email, "email": m.email, } for m in members ] def send_member_notification( db: Session, admin: User, *, user_id: int | None, notification_type: str, title: str, message: str, ) -> tuple[int, str, list[User]]: ntype = _normalize_member_notification_type(notification_type) if user_id is not None: target = db.get(User, user_id) if not target or target.role != UserRole.user or not target.is_active: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Member not found") targets = [target] recipient_label = target.full_name.strip() or target.email else: targets = ( db.query(User) .filter(User.role == UserRole.user, User.is_active.is_(True)) .order_by(User.id.asc()) .all() ) recipient_label = "All Members" if not targets: raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="No active members to notify") rows = [ MemberNotification( user_id=member.id, notification_type=ntype.value, title=title, message=message, sent_by_id=admin.id, ) for member in targets ] db.add_all(rows) db.commit() return len(rows), recipient_label, targets # --------------------------------------------------------------------------- # Content (Privacy / Terms / FAQ) # --------------------------------------------------------------------------- _DEFAULT_PAGES: dict[ContentPageSlug, tuple[str, str]] = { ContentPageSlug.privacy_policy: ( "Privacy Policy", "# Privacy Policy\n\nLast updated: April 28, 2026\n\n## Introduction\n\nYour privacy is important to us.", ), ContentPageSlug.terms_conditions: ( "Terms & Conditions", "# Terms & Conditions\n\nLast updated: April 28, 2026\n\n## Introduction\n\nPlease read these terms carefully.", ), } _CREATE_CONTENT_PAGES_PG = """ CREATE TABLE IF NOT EXISTS content_pages ( id SERIAL PRIMARY KEY, slug VARCHAR(64) NOT NULL UNIQUE, title VARCHAR(255) NOT NULL, body TEXT NOT NULL DEFAULT '', updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); """ _CREATE_FAQS_PG = """ CREATE TABLE IF NOT EXISTS faqs ( id SERIAL PRIMARY KEY, question VARCHAR(512) NOT NULL, answer TEXT NOT NULL, status VARCHAR(32) NOT NULL DEFAULT 'completed', view_count INTEGER NOT NULL DEFAULT 0, created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL, updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW() NOT NULL ); CREATE INDEX IF NOT EXISTS ix_faqs_id ON faqs (id); """ def sync_content_schema(engine: Engine) -> None: is_pg = engine.dialect.name == "postgresql" try: insp = inspect(engine) tables = set(insp.get_table_names()) except Exception: tables = set() if is_pg: with engine.begin() as conn: if "content_pages" not in tables: conn.execute(text(_CREATE_CONTENT_PAGES_PG)) if "faqs" not in tables: conn.execute(text(_CREATE_FAQS_PG)) else: from app import models_admin # noqa: F401 from app.database import Base if "content_pages" not in tables: Base.metadata.tables["content_pages"].create(bind=engine, checkfirst=True) if "faqs" not in tables: Base.metadata.tables["faqs"].create(bind=engine, checkfirst=True) def _parse_content_slug(raw: str) -> ContentPageSlug: normalized = raw.strip().lower().replace("-", "_") aliases = { "privacy": ContentPageSlug.privacy_policy, "privacy_policy": ContentPageSlug.privacy_policy, "terms": ContentPageSlug.terms_conditions, "terms_conditions": ContentPageSlug.terms_conditions, "terms_and_conditions": ContentPageSlug.terms_conditions, } if normalized not in aliases: raise ValueError("Invalid content_type. Use: privacy_policy, terms_conditions") return aliases[normalized] def _parse_faq_status(raw: str | None) -> FaqStatus: if not raw: return FaqStatus.completed normalized = raw.strip().lower() if normalized in ("completed", "complete", "published"): return FaqStatus.completed if normalized in ("draft", "pending"): return FaqStatus.draft raise ValueError("Invalid status. Use: completed or draft") def _format_updated(d) -> str: return "" if d is None else d.strftime("%d-%b-%Y") def serialize_content_page(page: ContentPage) -> dict: return { "content_type": page.slug.value if hasattr(page.slug, "value") else str(page.slug), "title": page.title, "content": page.body, "last_updated": page.updated_at, "last_updated_label": _format_updated(page.updated_at), } def serialize_faq(faq: Faq) -> dict: status_val = faq.status.value if hasattr(faq.status, "value") else str(faq.status) return { "id": faq.id, "question": faq.question, "answer": faq.answer, "status": status_val, "status_label": status_val.replace("_", " ").title(), "view_count": faq.view_count, "last_updated": faq.updated_at, "last_updated_label": _format_updated(faq.updated_at), "created_at": faq.created_at, } def get_or_create_content_page(db: Session, slug: ContentPageSlug) -> ContentPage: page = db.query(ContentPage).filter(ContentPage.slug == slug).first() if page: return page title, body = _DEFAULT_PAGES[slug] page = ContentPage(slug=slug, title=title, body=body) db.add(page) db.commit() db.refresh(page) return page def get_content_page(db: Session, content_type: str) -> ContentPage: return get_or_create_content_page(db, _parse_content_slug(content_type)) def save_content_page(db: Session, *, content_type: str, title: str, content: str) -> ContentPage: page = get_or_create_content_page(db, _parse_content_slug(content_type)) page.title = title.strip() page.body = content.strip() db.commit() db.refresh(page) return page def list_faqs( db: Session, *, search: str | None, skip: int, limit: int, published_only: bool = False, ) -> tuple[list[Faq], int]: q = db.query(Faq) if published_only: q = q.filter(Faq.status == FaqStatus.completed) if search and (term := search.strip()): pattern = f"%{term}%" q = q.filter(or_(Faq.question.ilike(pattern), Faq.answer.ilike(pattern))) total = q.count() rows = q.order_by(Faq.updated_at.desc()).offset(skip).limit(limit).all() return rows, total def create_faq(db: Session, *, question: str, answer: str, status: str | None) -> Faq: faq = Faq(question=question.strip(), answer=answer.strip(), status=_parse_faq_status(status)) db.add(faq) db.commit() db.refresh(faq) return faq def update_faq(db: Session, faq_id: int, *, question: str, answer: str, status: str | None) -> Faq | None: faq = db.get(Faq, faq_id) if not faq: return None faq.question = question.strip() faq.answer = answer.strip() faq.status = _parse_faq_status(status) db.commit() db.refresh(faq) return faq def delete_faq(db: Session, faq_id: int) -> bool: faq = db.get(Faq, faq_id) if not faq: return False db.delete(faq) db.commit() return True def get_faq(db: Session, faq_id: int) -> Faq | None: return db.get(Faq, faq_id)