804 lines
31 KiB
Python
804 lines
31 KiB
Python
"""上传案例：会员提交的结构化记录，与案例库栏目文章分开。"""
|
|
|
|
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
|