Spaces:
Running
Running
| # data_manager.py (修复版) | |
| import json | |
| import uuid | |
| from datetime import datetime | |
| from typing import List, Dict, Optional | |
| DATABASE_PATH = "database.json" | |
| def _load_database(db_path: str = DATABASE_PATH) -> List[Dict]: | |
| """ | |
| 内部函数：从指定的JSON文件加载数据库。 | |
| """ | |
| try: | |
| # 修改点：这里使用传入的 db_path，而不是全局变量 | |
| with open(db_path, "r", encoding="utf-8") as f: | |
| return json.load(f) | |
| except FileNotFoundError: | |
| return [] | |
| def _save_database(data: List[Dict], db_path: str = DATABASE_PATH) -> None: | |
| """ | |
| 内部函数：将数据保存到指定的JSON文件。 | |
| """ | |
| # 修改点：这里使用传入的 db_path | |
| with open(db_path, "w", encoding="utf-8") as f: | |
| json.dump(data, f, indent=4, ensure_ascii=False) | |
| def get_all_items(db_path: str = DATABASE_PATH) -> List[Dict]: | |
| """ | |
| 读取并返回指定数据库中所有物品的元数据。 | |
| """ | |
| return _load_database(db_path=db_path) | |
| def add_item( | |
| vec_id: int, | |
| description: str, | |
| location: str, | |
| category: str, | |
| image_filename: str, | |
| tags: List[str], | |
| contact: str, | |
| item_type: str, | |
| db_path: str = DATABASE_PATH, | |
| ) -> Dict: | |
| """ | |
| 向指定的数据库中添加一条新的物品记录。 | |
| """ | |
| # 这里之前报错，因为 _load_database 不接受参数。现在修好了。 | |
| items = _load_database(db_path=db_path) | |
| new_item = { | |
| "id": str(uuid.uuid4()), | |
| "vec_id": vec_id, | |
| "description": description, | |
| "location": location, | |
| "category": category, | |
| "image_filename": image_filename, | |
| "timestamp": datetime.utcnow().isoformat() + "Z", | |
| "tags": tags, | |
| "contact": contact, | |
| "item_type": item_type, | |
| } | |
| items.append(new_item) | |
| # 这里之前也会报错，现在修好了。 | |
| _save_database(items, db_path=db_path) | |
| return new_item | |
| def filter_items( | |
| location: Optional[str] = None, | |
| category: Optional[str] = None, | |
| date_range_start: Optional[str] = None, | |
| date_range_end: Optional[str] = None, | |
| db_path: str = DATABASE_PATH, | |
| ) -> List[Dict]: | |
| """ | |
| 根据一个或多个筛选条件从指定数据库过滤物品。 | |
| """ | |
| items = _load_database(db_path=db_path) | |
| filtered_items = [] | |
| start_date = ( | |
| datetime.fromisoformat(date_range_start.replace("Z", "+00:00")) | |
| if date_range_start | |
| else None | |
| ) | |
| end_date = ( | |
| datetime.fromisoformat(date_range_end.replace("Z", "+00:00")) | |
| if date_range_end | |
| else None | |
| ) | |
| for item in items: | |
| if location and item.get("location") != location: | |
| continue | |
| if category and item.get("category") != category: | |
| continue | |
| item_date = datetime.fromisoformat(item.get("timestamp").replace("Z", "+00:00")) | |
| if start_date and item_date < start_date: | |
| continue | |
| if end_date and item_date > end_date: | |
| continue | |
| filtered_items.append(item) | |
| return filtered_items | |
| def get_item_by_vec_id(vec_id: int, db_path: str = DATABASE_PATH) -> Optional[Dict]: | |
| """ | |
| 通过vec_id从指定数据库快速查找对应的物品元数据。 | |
| """ | |
| items = _load_database(db_path=db_path) | |
| for item in items: | |
| if item.get("vec_id") == vec_id: | |
| return item | |
| return None | |
| def get_next_vec_id(db_path: str = DATABASE_PATH) -> int: | |
| """ | |
| 获取指定数据库下一个可用的 vec_id。 | |
| """ | |
| items = _load_database(db_path=db_path) | |
| return len(items) |