"""上传案例：会员提交的结构化记录，与案例库栏目文章分开。""" import json from datetime import date, datetime, timezone from decimal import Decimal, InvalidOperation from pathlib import Path from sqlalchemy import Date, DateTime, Integer, String, Text, or_, select from sqlalchemy.orm import Mapped, mapped_column from app.db import Base PHOTO_MAX_BYTES = 5 * 1024 * 1024 PAGE_SIZES = (10, 20, 100) KIND_LABELS = { "residential": "住宅", "commercial": "商业", "office": "办公", "industrial": "工业", } OPEN_KINDS = frozenset({"residential", "commercial", "office", "industrial"}) CASE_TYPES = ("买卖成交", "拍卖成交", "月租金成交", "评估案例") STRUCTURES = ("钢混", "砖混", "混合", "钢结构") DECORATIONS = ("毛坯", "简易", "中档", "高档", "豪华") TOTAL_FLOORS = tuple(str(i) for i in range(1, 61)) FLOOR_NOS = ("-4", "-3", "-2", "-1") + TOTAL_FLOORS PURPOSES = ( "普通住宅", "公寓", "酒店式公寓", "独立别墅", "双拼别墅", "联排别墅", "停车位", "地下室", "其他", ) LAYOUT_STRUCTURES = ("平面", "复式", "跃式", "错层", "LOFT") OFFICE_LAYOUT_STRUCTURES = ("办公楼（中间过道两边办公室）", "写字楼（大开间）") LAYOUTS = ( "单间", "单身公寓", "一室一厅", "两室一厅", "两室两厅", "三室一厅", "三室两厅", "四室两厅", "五室两厅", "六室两厅", "七室及以上", ) COMMERCIAL_LAYOUTS = ("大开间", "隔档") STREET_FRONTAGES = ("单面临街", "双面临街", "三面临街", "十字路口拐角", "不临街（商场或市场内铺面）") FOOT_TRAFFICS = ("小", "较小", "一般", "较大", "大") SHOP_TYPES = ("住宅底商", "商业街商铺", "临街门面", "购物中心/百货内铺", "酒店宾馆内铺", "其他") BUSINESS_FORMATS = ( "空置", "餐饮", "服饰鞋帽箱包", "休闲娱乐", "美容美发", "服务业", "百货", "酒店宾馆", "家居建材", "酒吧", "汽车维修美容", "其他", ) ORIENTATIONS = ("东", "南", "西", "北", "东南", "西南", "东北", "西北", "南北", "东西") VALUE_INCLUDES = ("装饰装修", "家具用具", "停车位", "花园", "地下室") INDUSTRIAL_FLOORS = tuple(str(i) for i in range(1, 9)) INDUSTRIAL_LAYOUTS = ("开间",) INDUSTRIAL_DIRECTIONS = ("南北", "东西") ROAD_FRONTAGES = ("主干道", "次干道", "街巷") INDUSTRY_TYPES = ("食品饮料加工", "机械制造", "水泥厂", "热力公司", "其他") INDUSTRIAL_VALUE_INCLUDES = ("装饰装修", "家具用具") _REGIONS: list | None = None def _mysql_args() -> dict: return {"mysql_charset": "utf8mb4", "mysql_collate": "utf8mb4_unicode_ci"} def _now() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) class UploadedCase(Base): __tablename__ = "uploaded_case" __table_args__ = _mysql_args() id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) kind: Mapped[str] = mapped_column(String(32), nullable=False, index=True) member_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True) member_email: Mapped[str] = mapped_column(String(191), nullable=False, default="") case_type: Mapped[str] = mapped_column(String(32), nullable=False) province: Mapped[str] = mapped_column(String(64), nullable=False) city: Mapped[str] = mapped_column(String(64), nullable=False) district: Mapped[str] = mapped_column(String(64), nullable=False) estate_name: Mapped[str] = mapped_column(String(200), nullable=False) structure: Mapped[str] = mapped_column(String(32), nullable=False) case_date: Mapped[date] = mapped_column(Date, nullable=False) location: Mapped[str] = mapped_column(String(500), nullable=False) built_year: Mapped[int] = mapped_column(Integer, nullable=False) total_floors: Mapped[int] = mapped_column(Integer, nullable=False) floor_no: Mapped[str] = mapped_column(String(8), nullable=False) purpose: Mapped[str] = mapped_column(String(32), nullable=False) layout_structure: Mapped[str] = mapped_column(String(32), nullable=False) decoration: Mapped[str] = mapped_column(String(32), nullable=False) value_includes: Mapped[str] = mapped_column(Text, nullable=False, default="[]") area: Mapped[str] = mapped_column(String(32), nullable=False) layout: Mapped[str] = mapped_column(String(32), nullable=False) orientation: Mapped[str] = mapped_column(String(16), nullable=False) unit_price: Mapped[str] = mapped_column(String(32), nullable=False) remark: Mapped[str] = mapped_column(Text, nullable=False, default="") institution_name: Mapped[str] = mapped_column(String(200), nullable=False) street_frontage: Mapped[str] = mapped_column(String(64), nullable=False, default="") foot_traffic: Mapped[str] = mapped_column(String(32), nullable=False, default="") shop_type: Mapped[str] = mapped_column(String(64), nullable=False, default="") business_formats: Mapped[str] = mapped_column(Text, nullable=False, default="[]") heading: Mapped[str] = mapped_column(String(16), nullable=False, default="") road_frontage: Mapped[str] = mapped_column(String(32), nullable=False, default="") industry_type: Mapped[str] = mapped_column(String(32), nullable=False, default="") created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False, default=_now, index=True) class UploadedCasePhoto(Base): __tablename__ = "uploaded_case_photo" __table_args__ = _mysql_args() id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) case_id: Mapped[int] = mapped_column(Integer, nullable=False, index=True) file_id: Mapped[int] = mapped_column(Integer, nullable=False) position: Mapped[int] = mapped_column(Integer, nullable=False, default=0) def regions() -> list: global _REGIONS if _REGIONS is None: path = Path(__file__).resolve().parents[1] / "data" / "china_regions.json" _REGIONS = json.loads(path.read_text(encoding="utf-8")) return _REGIONS def built_years(today: date | None = None) -> list[int]: year = (today or date.today()).year return list(range(1980, year + 1)) def ensure_uploaded_case_columns(db) -> None: bind = db.get_bind() if bind.dialect.name != "mysql": return from sqlalchemy import inspect, text names = {column["name"] for column in inspect(bind).get_columns("uploaded_case")} statements = [] if "street_frontage" not in names: statements.append( "ALTER TABLE uploaded_case ADD COLUMN street_frontage VARCHAR(64) NOT NULL DEFAULT ''" ) if "foot_traffic" not in names: statements.append( "ALTER TABLE uploaded_case ADD COLUMN foot_traffic VARCHAR(32) NOT NULL DEFAULT ''" ) if "shop_type" not in names: statements.append("ALTER TABLE uploaded_case ADD COLUMN shop_type VARCHAR(64) NOT NULL DEFAULT ''") if "business_formats" not in names: statements.append("ALTER TABLE uploaded_case ADD COLUMN business_formats TEXT NULL") if "heading" not in names: statements.append("ALTER TABLE uploaded_case ADD COLUMN heading VARCHAR(16) NOT NULL DEFAULT ''") if "road_frontage" not in names: statements.append("ALTER TABLE uploaded_case ADD COLUMN road_frontage VARCHAR(32) NOT NULL DEFAULT ''") if "industry_type" not in names: statements.append("ALTER TABLE uploaded_case ADD COLUMN industry_type VARCHAR(32) NOT NULL DEFAULT ''") for statement in statements: db.execute(text(statement)) if "business_formats" not in names: db.execute(text("UPDATE uploaded_case SET business_formats = '[]' WHERE business_formats IS NULL")) if statements: db.commit() def commercial_options(today: date | None = None) -> dict: return { "caseTypes": list(CASE_TYPES), "structures": list(STRUCTURES), "builtYears": built_years(today), "totalFloors": list(TOTAL_FLOORS), "floorNos": list(FLOOR_NOS), "layouts": list(COMMERCIAL_LAYOUTS), "decorations": list(DECORATIONS), "streetFrontages": list(STREET_FRONTAGES), "orientations": list(ORIENTATIONS), "footTraffics": list(FOOT_TRAFFICS), "shopTypes": list(SHOP_TYPES), "businessFormats": list(BUSINESS_FORMATS), "valueIncludes": list(VALUE_INCLUDES), } def office_options(today: date | None = None) -> dict: return { "caseTypes": list(CASE_TYPES), "structures": list(STRUCTURES), "builtYears": built_years(today), "totalFloors": list(TOTAL_FLOORS), "floorNos": list(FLOOR_NOS), "layoutStructures": list(OFFICE_LAYOUT_STRUCTURES), "decorations": list(DECORATIONS), "valueIncludes": list(VALUE_INCLUDES), "orientations": list(ORIENTATIONS), } def residential_options(today: date | None = None) -> dict: return { "caseTypes": list(CASE_TYPES), "structures": list(STRUCTURES), "builtYears": built_years(today), "totalFloors": list(TOTAL_FLOORS), "floorNos": list(FLOOR_NOS), "purposes": list(PURPOSES), "layoutStructures": list(LAYOUT_STRUCTURES), "decorations": list(DECORATIONS), "valueIncludes": list(VALUE_INCLUDES), "layouts": list(LAYOUTS), "orientations": list(ORIENTATIONS), } def industrial_options(today: date | None = None) -> dict: return { "caseTypes": list(CASE_TYPES), "structures": list(STRUCTURES), "builtYears": built_years(today), "totalFloors": list(INDUSTRIAL_FLOORS), "layouts": list(INDUSTRIAL_LAYOUTS), "headings": list(INDUSTRIAL_DIRECTIONS), "decorations": list(DECORATIONS), "roadFrontages": list(ROAD_FRONTAGES), "orientations": list(INDUSTRIAL_DIRECTIONS), "industryTypes": list(INDUSTRY_TYPES), "valueIncludes": list(INDUSTRIAL_VALUE_INCLUDES), } def _choice(value, options: tuple[str, ...] | list, label: str) -> str: text = str(value or "").strip() if text not in options: raise ValueError(f"{label}不正确") return text def _text(value, label: str, limit: int) -> str: text = str(value or "").strip() if not text: raise ValueError(f"请填写{label}") if len(text) > limit: raise ValueError(f"{label}过长") return text def _number(value, label: str) -> str: raw = str(value if value is not None else "").strip() if not raw: raise ValueError(f"请填写{label}") try: number = Decimal(raw) except InvalidOperation as exc: raise ValueError(f"{label}须为数字") from exc if not number.is_finite(): raise ValueError(f"{label}须为数字") return format(number, "f") def _region(province: str, city: str, district: str) -> tuple[str, str, str]: province = str(province or "").strip() city = str(city or "").strip() district = str(district or "").strip() for item in regions(): if item.get("name") != province: continue for child in item.get("sub") or []: if child.get("name") != city: continue names = [row.get("name") for row in child.get("sub") or []] if district not in names: raise ValueError("区县不正确") return province, city, district raise ValueError("城市不正确") raise ValueError("省份不正确") def _includes(value, options: tuple[str, ...] = VALUE_INCLUDES) -> list[str]: if value is None or value == "" or value == []: return [] if not isinstance(value, list): raise ValueError("价值包含不正确") chosen = [] for item in value: text = str(item or "").strip() if text not in options or text in chosen: raise ValueError("价值包含不正确") chosen.append(text) if len(chosen) > 3: raise ValueError("价值包含最多 3 项") return chosen def _formats(value) -> list[str]: if value is None or value == "" or value == []: raise ValueError("请选择经营业态") if not isinstance(value, list): raise ValueError("经营业态不正确") chosen = [] for item in value: text = str(item or "").strip() if text not in BUSINESS_FORMATS or text in chosen: raise ValueError("经营业态不正确") chosen.append(text) if not chosen: raise ValueError("请选择经营业态") if len(chosen) > 3: raise ValueError("经营业态最多 3 项") return chosen def _photos(db, member_id: int, file_ids) -> list[int]: from app.models.content import AdminFile if not isinstance(file_ids, list) or not file_ids: raise ValueError("请上传位置图及案例照片") ids: list[int] = [] for raw in file_ids: try: file_id = int(raw) except (TypeError, ValueError) as exc: raise ValueError("位置图及案例照片不正确") from exc row = db.get(AdminFile, file_id) if row is None or row.bucket != "private" or row.created_by != member_id: raise ValueError("位置图及案例照片不正确") if row.size > PHOTO_MAX_BYTES: raise ValueError("位置图及案例照片须小于 5MB") if file_id not in ids: ids.append(file_id) return ids def _residential(db, member_id: int, payload: dict, today: date | None = None) -> dict: extra = set(payload) - { "kind", "caseType", "province", "city", "district", "estateName", "structure", "caseDate", "location", "builtYear", "totalFloors", "floorNo", "purpose", "layoutStructure", "decoration", "valueIncludes", "area", "layout", "orientation", "unitPrice", "remark", "institutionName", "photoFileIds", } if extra: raise ValueError("包含住宅没有的字段") province, city, district = _region(payload.get("province"), payload.get("city"), payload.get("district")) raw_date = str(payload.get("caseDate") or "").strip() try: case_date = datetime.strptime(raw_date, "%Y-%m-%d").date() except ValueError as exc: raise ValueError("案例时间不正确") from exc year = _choice(str(payload.get("builtYear") or "").strip(), [str(item) for item in built_years(today)], "建筑年代") remark = str(payload.get("remark") or "").strip() if remark and not 1 <= len(remark) <= 1000: raise ValueError("备注长度须在 1 到 1000") return { "case_type": _choice(payload.get("caseType"), CASE_TYPES, "案例类型"), "province": province, "city": city, "district": district, "estate_name": _text(payload.get("estateName"), "楼盘名称", 200), "structure": _choice(payload.get("structure"), STRUCTURES, "建筑结构"), "case_date": case_date, "location": _text(payload.get("location"), "坐落", 500), "built_year": int(year), "total_floors": int(_choice(str(payload.get("totalFloors") or "").strip(), TOTAL_FLOORS, "总层数")), "floor_no": _choice(str(payload.get("floorNo") or "").strip(), FLOOR_NOS, "所在层数"), "purpose": _choice(payload.get("purpose"), PURPOSES, "用途"), "layout_structure": _choice(payload.get("layoutStructure"), LAYOUT_STRUCTURES, "户型结构"), "decoration": _choice(payload.get("decoration"), DECORATIONS, "装修档次"), "value_includes": _includes(payload.get("valueIncludes")), "area": _number(payload.get("area"), "建筑面积"), "layout": _choice(payload.get("layout"), LAYOUTS, "户型"), "orientation": _choice(payload.get("orientation"), ORIENTATIONS, "朝向"), "unit_price": _number(payload.get("unitPrice"), "单价"), "remark": remark, "institution_name": _text(payload.get("institutionName"), "提交机构名称", 200), "photo_ids": _photos(db, member_id, payload.get("photoFileIds")), } def _commercial(db, member_id: int, payload: dict, today: date | None = None) -> dict: extra = set(payload) - { "kind", "caseType", "province", "city", "district", "estateName", "structure", "caseDate", "location", "builtYear", "totalFloors", "floorNo", "layout", "decoration", "streetFrontage", "area", "orientation", "footTraffic", "shopType", "businessFormats", "unitPrice", "valueIncludes", "remark", "institutionName", "photoFileIds", } if extra: raise ValueError("包含商业没有的字段") province, city, district = _region(payload.get("province"), payload.get("city"), payload.get("district")) raw_date = str(payload.get("caseDate") or "").strip() try: case_date = datetime.strptime(raw_date, "%Y-%m-%d").date() except ValueError as exc: raise ValueError("案例时间不正确") from exc year = _choice(str(payload.get("builtYear") or "").strip(), [str(item) for item in built_years(today)], "建筑年代") remark = str(payload.get("remark") or "").strip() if remark and not 1 <= len(remark) <= 1000: raise ValueError("备注长度须在 1 到 1000") return { "case_type": _choice(payload.get("caseType"), CASE_TYPES, "案例类型"), "province": province, "city": city, "district": district, "estate_name": _text(payload.get("estateName"), "楼盘名称", 200), "structure": _choice(payload.get("structure"), STRUCTURES, "建筑结构"), "case_date": case_date, "location": _text(payload.get("location"), "坐落", 500), "built_year": int(year), "total_floors": int(_choice(str(payload.get("totalFloors") or "").strip(), TOTAL_FLOORS, "总层数")), "floor_no": _choice(str(payload.get("floorNo") or "").strip(), FLOOR_NOS, "所在层数"), "purpose": "", "layout_structure": "", "decoration": _choice(payload.get("decoration"), DECORATIONS, "装修档次"), "value_includes": _includes(payload.get("valueIncludes")), "area": _number(payload.get("area"), "建筑面积"), "layout": _choice(payload.get("layout"), COMMERCIAL_LAYOUTS, "户型"), "orientation": _choice(payload.get("orientation"), ORIENTATIONS, "朝向"), "street_frontage": _choice(payload.get("streetFrontage"), STREET_FRONTAGES, "临街状况"), "foot_traffic": _choice(payload.get("footTraffic"), FOOT_TRAFFICS, "人流量"), "shop_type": _choice(payload.get("shopType"), SHOP_TYPES, "商铺类型"), "business_formats": _formats(payload.get("businessFormats")), "unit_price": _number(payload.get("unitPrice"), "单价"), "remark": remark, "institution_name": _text(payload.get("institutionName"), "提交机构名称", 200), "photo_ids": _photos(db, member_id, payload.get("photoFileIds")), } def _office(db, member_id: int, payload: dict, today: date | None = None) -> dict: extra = set(payload) - { "kind", "caseType", "province", "city", "district", "estateName", "structure", "caseDate", "location", "builtYear", "totalFloors", "floorNo", "layoutStructure", "decoration", "valueIncludes", "area", "orientation", "unitPrice", "remark", "institutionName", "photoFileIds", } if extra: raise ValueError("包含办公没有的字段") province, city, district = _region(payload.get("province"), payload.get("city"), payload.get("district")) raw_date = str(payload.get("caseDate") or "").strip() try: case_date = datetime.strptime(raw_date, "%Y-%m-%d").date() except ValueError as exc: raise ValueError("案例时间不正确") from exc year = _choice(str(payload.get("builtYear") or "").strip(), [str(item) for item in built_years(today)], "建筑年代") remark = str(payload.get("remark") or "").strip() if remark and not 1 <= len(remark) <= 1000: raise ValueError("备注长度须在 1 到 1000") return { "case_type": _choice(payload.get("caseType"), CASE_TYPES, "案例类型"), "province": province, "city": city, "district": district, "estate_name": _text(payload.get("estateName"), "楼盘名称", 200), "structure": _choice(payload.get("structure"), STRUCTURES, "建筑结构"), "case_date": case_date, "location": _text(payload.get("location"), "坐落", 500), "built_year": int(year), "total_floors": int(_choice(str(payload.get("totalFloors") or "").strip(), TOTAL_FLOORS, "总层数")), "floor_no": _choice(str(payload.get("floorNo") or "").strip(), FLOOR_NOS, "所在层数"), "purpose": "", "layout_structure": _choice(payload.get("layoutStructure"), OFFICE_LAYOUT_STRUCTURES, "户型结构"), "decoration": _choice(payload.get("decoration"), DECORATIONS, "装修档次"), "value_includes": _includes(payload.get("valueIncludes")), "area": _number(payload.get("area"), "建筑面积"), "layout": "", "orientation": _choice(payload.get("orientation"), ORIENTATIONS, "朝向"), "unit_price": _number(payload.get("unitPrice"), "单价"), "remark": remark, "institution_name": _text(payload.get("institutionName"), "提交机构名称", 200), "photo_ids": _photos(db, member_id, payload.get("photoFileIds")), } def _industrial(db, member_id: int, payload: dict, today: date | None = None) -> dict: extra = set(payload) - { "kind", "caseType", "province", "city", "district", "estateName", "structure", "caseDate", "location", "builtYear", "totalFloors", "layout", "heading", "decoration", "roadFrontage", "area", "orientation", "industryType", "unitPrice", "valueIncludes", "remark", "institutionName", "photoFileIds", } if extra: raise ValueError("包含工业没有的字段") province, city, district = _region(payload.get("province"), payload.get("city"), payload.get("district")) raw_date = str(payload.get("caseDate") or "").strip() try: case_date = datetime.strptime(raw_date, "%Y-%m-%d").date() except ValueError as exc: raise ValueError("案例时间不正确") from exc year = _choice(str(payload.get("builtYear") or "").strip(), [str(item) for item in built_years(today)], "建筑年代") remark = str(payload.get("remark") or "").strip() if remark and not 1 <= len(remark) <= 1000: raise ValueError("备注长度须在 1 到 1000") return { "case_type": _choice(payload.get("caseType"), CASE_TYPES, "案例类型"), "province": province, "city": city, "district": district, "estate_name": _text(payload.get("estateName"), "楼盘名称", 200), "structure": _choice(payload.get("structure"), STRUCTURES, "建筑结构"), "case_date": case_date, "location": _text(payload.get("location"), "坐落", 500), "built_year": int(year), "total_floors": int(_choice(str(payload.get("totalFloors") or "").strip(), INDUSTRIAL_FLOORS, "总层数")), "floor_no": "", "purpose": "", "layout_structure": "", "decoration": _choice(payload.get("decoration"), DECORATIONS, "装修档次"), "value_includes": _includes(payload.get("valueIncludes"), INDUSTRIAL_VALUE_INCLUDES), "area": _number(payload.get("area"), "建筑面积"), "layout": _choice(payload.get("layout"), INDUSTRIAL_LAYOUTS, "户型"), "heading": _choice(payload.get("heading"), INDUSTRIAL_DIRECTIONS, "走向"), "road_frontage": _choice(payload.get("roadFrontage"), ROAD_FRONTAGES, "临路状况"), "orientation": _choice(payload.get("orientation"), INDUSTRIAL_DIRECTIONS, "朝向"), "industry_type": _choice(payload.get("industryType"), INDUSTRY_TYPES, "工业类型"), "unit_price": _number(payload.get("unitPrice"), "单价"), "remark": remark, "institution_name": _text(payload.get("institutionName"), "提交机构名称", 200), "photo_ids": _photos(db, member_id, payload.get("photoFileIds")), } def create_uploaded_case(db, member, payload: dict, today: date | None = None) -> UploadedCase: if not isinstance(payload, dict): raise ValueError("提交内容不正确") kind = str(payload.get("kind") or "").strip() if kind not in KIND_LABELS: raise ValueError("案例种类不正确") if kind not in OPEN_KINDS: raise ValueError("该类型尚未开放") parsers = { "residential": _residential, "commercial": _commercial, "office": _office, "industrial": _industrial, } fields = parsers[kind](db, member.id, payload, today) row = UploadedCase( kind=kind, member_id=member.id, member_email=member.email, case_type=fields["case_type"], province=fields["province"], city=fields["city"], district=fields["district"], estate_name=fields["estate_name"], structure=fields["structure"], case_date=fields["case_date"], location=fields["location"], built_year=fields["built_year"], total_floors=fields["total_floors"], floor_no=fields["floor_no"], purpose=fields["purpose"], layout_structure=fields["layout_structure"], decoration=fields["decoration"], value_includes=json.dumps(fields["value_includes"], ensure_ascii=False), area=fields["area"], layout=fields["layout"], orientation=fields["orientation"], unit_price=fields["unit_price"], remark=fields["remark"], institution_name=fields["institution_name"], street_frontage=fields.get("street_frontage", ""), foot_traffic=fields.get("foot_traffic", ""), shop_type=fields.get("shop_type", ""), business_formats=json.dumps(fields.get("business_formats", []), ensure_ascii=False), heading=fields.get("heading", ""), road_frontage=fields.get("road_frontage", ""), industry_type=fields.get("industry_type", ""), created_at=_now(), ) db.add(row) db.flush() for index, file_id in enumerate(fields["photo_ids"]): db.add(UploadedCasePhoto(case_id=row.id, file_id=file_id, position=index)) db.commit() db.refresh(row) return row def _address(row: UploadedCase) -> str: return f"{row.province} {row.city} {row.district}" def _list_item(row: UploadedCase) -> dict: return { "id": row.id, "kind": row.kind, "kindLabel": KIND_LABELS.get(row.kind, row.kind), "estateName": row.estate_name, "address": _address(row), "caseDate": row.case_date.isoformat() if row.case_date else "", "institutionName": row.institution_name, "memberEmail": row.member_email, "createdAt": row.created_at.isoformat(timespec="seconds") if row.created_at else "", } def _photos_payload(db, case_id: int) -> list[dict]: from app.models.content import AdminFile from app.services.storage import private_object_url links = db.scalars( select(UploadedCasePhoto) .where(UploadedCasePhoto.case_id == case_id) .order_by(UploadedCasePhoto.position.asc(), UploadedCasePhoto.id.asc()) ).all() items = [] for link in links: file_row = db.get(AdminFile, link.file_id) if file_row is None or file_row.bucket != "private": continue url, expires_at = private_object_url(file_row.object_key) items.append( { "id": file_row.id, "filename": file_row.filename, "contentType": file_row.content_type or "", "url": url, "expiresAt": expires_at.isoformat(), } ) return items def _like_contains(value: str) -> str: escaped = value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") return f"%{escaped}%" def list_uploaded_cases(db, *, kind: str, current: int, size: int, q: str | None = None) -> dict | None: if kind not in KIND_LABELS: return None page_size = size if size in PAGE_SIZES else 10 page = current if current >= 1 else 1 filters = [UploadedCase.kind == kind] keyword = (q or "").strip()[:100] if keyword: needle = _like_contains(keyword) filters.append( or_( UploadedCase.estate_name.like(needle, escape="\\"), UploadedCase.province.like(needle, escape="\\"), UploadedCase.city.like(needle, escape="\\"), UploadedCase.district.like(needle, escape="\\"), UploadedCase.institution_name.like(needle, escape="\\"), UploadedCase.member_email.like(needle, escape="\\"), ) ) total = len(db.scalars(select(UploadedCase.id).where(*filters)).all()) rows = db.scalars( select(UploadedCase) .where(*filters) .order_by(UploadedCase.created_at.desc(), UploadedCase.id.desc()) .offset((page - 1) * page_size) .limit(page_size) ).all() return { "kind": kind, "kindLabel": KIND_LABELS[kind], "items": [_list_item(row) for row in rows], "total": total, "current": page, "size": page_size, } def uploaded_case_detail(db, case_id: int) -> dict | None: row = db.get(UploadedCase, case_id) if row is None: return None body = _list_item(row) body.update( { "caseType": row.case_type, "province": row.province, "city": row.city, "district": row.district, "structure": row.structure, "location": row.location, "builtYear": row.built_year, "totalFloors": row.total_floors, "decoration": row.decoration, "valueIncludes": json.loads(row.value_includes or "[]"), "area": row.area, "orientation": row.orientation, "unitPrice": row.unit_price, "remark": row.remark, "photos": _photos_payload(db, row.id), } ) if row.kind != "industrial": body["floorNo"] = row.floor_no if row.kind in ("residential", "commercial", "industrial"): body["layout"] = row.layout if row.kind == "residential": body["purpose"] = row.purpose body["layoutStructure"] = row.layout_structure if row.kind == "office": body["layoutStructure"] = row.layout_structure if row.kind == "commercial": body["streetFrontage"] = row.street_frontage body["footTraffic"] = row.foot_traffic body["shopType"] = row.shop_type body["businessFormats"] = json.loads(row.business_formats or "[]") if row.kind == "industrial": body["heading"] = row.heading body["roadFrontage"] = row.road_frontage body["industryType"] = row.industry_type return body