services.inventory_manager
源文件
services/inventory_manager.py· 由scripts/gen_api_docs.py自动生成,请勿手改
模块说明:
库存管理数据层 — 机库 CRUD / 物品入库 / 加权平均成本 / 移动
函数
_default_db
def _default_db() -> DatabaseManager惰性获取 DatabaseManager(经容器,消除模块级单例双轨)。
定义行:25
init_db
def init_db()⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:72
get_hangars
def get_hangars() -> list[dict]⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:89
create_hangar
def create_hangar(name: str, solar_system_id: int | None=None) -> int⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:107
rename_hangar
def rename_hangar(hangar_id: int, name: str) -> bool⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:117
update_hangar_system
def update_hangar_system(hangar_id: int, solar_system_id: int | None) -> bool设置机库所在星系(None 清除)。
定义行:124
get_hangar_system_id
def get_hangar_system_id(hangar_id: int | None, *, _db: DatabaseManager | None=None) -> int | None读取机库所在星系的 solar_system_id(无机库/未设置返回 None)。
定义行:132
get_hangar_name
def get_hangar_name(hangar_id: int | None) -> str读取机库名称(无机库/未设置返回空串)。
定义行:142
get_default_mat_hangar_and_system
def get_default_mat_hangar_and_system() -> tuple[int | None, int | None]返回 (默认材料机库 id, 其所在星系 id)。settings 未配置/读取失败 → (None, None)。
定义行:154
get_default_mat_hangar_system_id
def get_default_mat_hangar_system_id() -> int | None从 settings.json 默认材料机库带出星系(无计划上下文的 SCI 依据)。
定义行:173
update_hangar_config
def update_hangar_config(hangar_id: int, facility_type: str | None, facility_tax: float | None, rigs: list[int] | None) -> bool更新机库工业配置(设施类型/设施税/改件)。rigs 存 JSON 数组。
定义行:178
get_hangar_config
def get_hangar_config(hangar_id: int | None) -> dict读取机库工业配置 {facility_type, facility_tax, rigs: list[int]};无机库/未配置返回默认。
定义行:194
hangar_references
def hangar_references(hangar_id: int) -> list[str]返回引用该机库的位置描述,如 ['默认材料机库', '3 条计划的材料机库']。
定义行:223
repoint_hangar_references
def repoint_hangar_references(old_id: int, new_id: int) -> int把指向 old_id 的引用改指到 new_id,返回改动处数。
定义行:252
delete_hangar
def delete_hangar(hangar_id: int, *, repoint_to: int | None=None) -> bool删除机库及其物品(蓝图/物品由 ON DELETE CASCADE 连带删除)。
定义行:284
get_items
def get_items(hangar_id: int, *, include_derived: bool=True, need_ids: set[int] | None=None) -> list[dict]机库物品列表。
定义行:299
get_item_price
def get_item_price(type_id: int) -> float | None⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:401
get_inventory_cost_map
def get_inventory_cost_map(_db: DatabaseManager | None=None) -> dict[int, tuple[int, float]]跨机库汇总各物品库存数量与加权平均成本(成本按数量加权)。
定义行:409
add_item
def add_item(hangar_id: int, type_id: int, quantity: int, cost_price: float=0, *, conn=None) -> int把 quantity 件物品加入机库,按加权平均成本合并。
定义行:440
set_item_quantity
def set_item_quantity(hangar_id: int, type_id: int, quantity: int, cost_price: float | None=None, *, conn=None) -> int全量同步:把 (hangar_id, type_id) 的数量设为 quantity。
定义行:478
update_cost_price
def update_cost_price(item_id: int, cost_price: float) -> bool直接覆盖该库存行的单位成本价(参数化 UPDATE,返回是否命中)。
定义行:528
get_hangar_stock
def get_hangar_stock(hangar_id: int) -> dict[int, int]单机库库存快照 {type_id: quantity}(quantity > 0 才计入)。
定义行:536
get_hangar_cost_map
def get_hangar_cost_map(hangar_id: int) -> dict[int, float]单机库物品成本快照 {type_id: 加权平均成本}。
定义行:553
deduct_item
def deduct_item(hangar_id: int, type_id: int, quantity: int, *, conn=None) -> int从机库扣减 quantity,返回实际扣减量。
定义行:569
remove_item
def remove_item(item_id: int) -> bool⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:605
update_quantity
def update_quantity(item_id: int, quantity: int) -> bool⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:612
move_items
def move_items(item_ids: list[int], to_hangar_id: int)⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:624
_move_item_between_hangars
def _move_item_between_hangars(src_hangar: int, type_id: int, target_hangar: int, *, conn: sqlite3.Connection | None=None) -> bool把源机库中 type_id 的物品按数量整体移到目标机库,返回是否移动。
定义行:656
_move_item_row
def _move_item_row(conn: sqlite3.Connection, src_hangar: int, type_id: int, target_hangar: int) -> bool_move_item_between_hangars 的实际搬运(在同一连接/事务内,由调用方提交)。
定义行:675
apply_inventory_import
def apply_inventory_import(hangar_id: int, data: list[tuple[int, int, float, int | None]], mode: str, targets: dict[int, int] | None=None, clear_missing: dict[int, int] | None=None) -> tuple[int, int]按导入数据应用库存变更,返回 (added, moved)。
定义行:718
get_total_value
def get_total_value(hangar_id: int, price_type: str='sell', discount: float=0) -> dict机库估值合计。
定义行:775
add_blueprint
def add_blueprint(hangar_id: int, blueprint_type_id: int, is_bpo: bool=True, me_level: int=0, te_level: int=0, runs: int=1, quantity: int=1, notes: str='', *, conn=None) -> int新增蓝图。conn 传入时在同一连接执行且不提交(由调用方统一事务)。
定义行:844
get_blueprints
def get_blueprints(hangar_id: int | None=None) -> list[dict]获取用户蓝图列表,可指定机库或全部
定义行:873
update_blueprint
def update_blueprint(bp_id: int, *, conn=None, **kwargs) -> bool更新蓝图属性,kwargs 可含 is_bpo, me_level, te_level, runs, quantity, notes
定义行:913
delete_blueprint
def delete_blueprint(bp_id: int, *, conn=None) -> bool删除蓝图。conn 传入时在同一连接执行且不提交(由调用方统一事务)。
定义行:938
delete_blueprints_batch
def delete_blueprints_batch(ids: list[int]) -> int批量删除蓝图,返回删除行数
定义行:960
move_blueprints_to_hangar
def move_blueprints_to_hangar(ids: list[int], hangar_id: int) -> int批量移动蓝图到目标机库
定义行:982
update_blueprints_batch
def update_blueprints_batch(ids: list[int], **kwargs) -> int批量更新蓝图属性(me_level, te_level, runs, cost_per_run 等)
定义行:993
get_blueprint_product_info
def get_blueprint_product_info(blueprint_type_id: int) -> dict | None获取蓝图的产物信息(名称、产量、制造时间)
定义行:1010
get_blueprint_product_info_batch
def get_blueprint_product_info_batch(bp_ids: list[int]) -> dict[int, dict]批量获取蓝图产物信息,返回 {blueprint_type_id: {product_type_id, product_name, product_quantity, base_time}}
定义行:1040
get_blueprint_materials_batch
def get_blueprint_materials_batch(bp_ids: list[int]) -> dict[int, list[tuple[int, int]]]批量获取蓝图材料,返回 {blueprint_type_id: [(material_type_id, quantity), ...]}
定义行:1071
check_blueprint_exists
def check_blueprint_exists(blueprint_type_id: int) -> bool检查用户蓝图库中是否已存在指定类型的蓝图
定义行:1092
get_blueprint_tech_levels
def get_blueprint_tech_levels()从 reference.db 获取各蓝图的科技等级
定义行:1100
get_blueprint_reaction_ids
def get_blueprint_reaction_ids() -> set[int]获取所有反应公式的 blueprint_type_id
定义行:1137
format_blueprint_status
def format_blueprint_status(statuses: Iterable[str]) -> str状态集合 → 显示串:按 BLUEPRINT_STATUSES 顺序以 · 连接;全未命中 → -。
定义行:1173
_in_batches
def _in_batches(type_ids: list[int]) -> Iterator[list[int]]⚠️ 待补 docstring
此函数暂无 docstring,欢迎补充。
定义行:1190
_stocked_type_ids
def _stocked_type_ids(conn: sqlite3.Connection, type_ids: list[int]) -> set[int]库中有成品:inventory_items.quantity > 0,全部机库合计(不限当前机库)。
定义行:1195
_listed_type_ids
def _listed_type_ids(conn: sqlite3.Connection, type_ids: list[int]) -> set[int]有挂单:open_orders.volume_remain > 0,买单卖单都算(没筛 is_buy)。
定义行:1210
get_stock_and_order_flags
def get_stock_and_order_flags(type_ids: Iterable[int], db: DatabaseManager | None=None) -> dict[int, tuple[bool, bool]]批量取 {type_id: (库中有, 有挂单)},只含命中项(未命中视为 (False, False))。
定义行:1225
get_production_state_flags
def get_production_state_flags(type_ids: Iterable[int], db: DatabaseManager | None=None) -> dict[int, frozenset[str]]批量取「这个产物现在处于什么状态」:{product_type_id: {标记…}}。
定义行:1251
get_blueprint_status_map
def get_blueprint_status_map(rows: Iterable[dict[str, Any]]) -> dict[int, str]批量取蓝图「状态」列显示串,返回 {user_blueprints.id: 显示串}。
定义行:1349