commit 544e4e072051bb547977f75a6c799682a058707c Author: sinmb79 Date: Fri Apr 3 09:08:08 2026 +0900 Initial CivilPlan MCP implementation diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..38cc898 --- /dev/null +++ b/.env.example @@ -0,0 +1,2 @@ +DATA_GO_KR_API_KEY= +VWORLD_API_KEY= diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..4c99638 --- /dev/null +++ b/.gitignore @@ -0,0 +1,12 @@ +.env +.venv/ +*.pyc +__pycache__/ +.pytest_cache/ +output/ +civilplan.db +civilplan_mcp/data/land_prices/* +!civilplan_mcp/data/land_prices/.gitkeep +!civilplan_mcp/data/land_prices/README.md +civilplan_mcp/data/.update_required_* +civilplan_mcp/data/update_log.json diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..8492c44 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 22B Labs + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..7567e60 --- /dev/null +++ b/README.md @@ -0,0 +1,130 @@ +# CivilPlan MCP + +LICENSE: MIT + +PHILOSOPHY: Hongik Ingan (홍익인간) - reduce inequality in access to expert planning knowledge. +This project is free to use, modify, and distribute. + +CivilPlan is a FastMCP server for Korean civil, infrastructure, and building project planning workflows. + +## Scope + +- FastMCP Streamable HTTP server on `http://127.0.0.1:8765/mcp` +- 19 planning tools across v1.0 and v1.1 +- Local JSON and SQLite bootstrap data +- Excel, DOCX, SVG, and DXF output generators +- Scheduled update scaffolding for wage, waste, and standard cost references + +## Tool Catalog + +### v1.0 + +1. `civilplan_parse_project` +2. `civilplan_get_legal_procedures` +3. `civilplan_get_phase_checklist` +4. `civilplan_evaluate_impact_assessments` +5. `civilplan_estimate_quantities` +6. `civilplan_get_unit_prices` +7. `civilplan_generate_boq_excel` +8. `civilplan_generate_investment_doc` +9. `civilplan_generate_schedule` +10. `civilplan_generate_svg_drawing` +11. `civilplan_get_applicable_guidelines` +12. `civilplan_fetch_guideline_summary` +13. `civilplan_select_bid_type` +14. `civilplan_estimate_waste_disposal` + +### v1.1 + +15. `civilplan_query_land_info` +16. `civilplan_analyze_feasibility` +17. `civilplan_validate_against_benchmark` +18. `civilplan_generate_budget_report` +19. `civilplan_generate_dxf_drawing` + +## Quick Start + +```bash +python -m venv .venv +.venv\Scripts\activate +pip install -r requirements.txt +copy .env.example .env +python server.py +``` + +The server starts on `127.0.0.1:8765` and exposes the MCP endpoint at `/mcp`. + +## Environment Variables + +```env +DATA_GO_KR_API_KEY= +VWORLD_API_KEY= +``` + +- `DATA_GO_KR_API_KEY`: public data portal API key used for benchmark probing and future official integrations +- `VWORLD_API_KEY`: VWorld API key used for address-to-PNU lookup and cadastral queries + +## Client Connection + +### Claude Desktop + +```json +{ + "mcpServers": { + "civilplan": { + "command": "mcp-remote", + "args": ["http://127.0.0.1:8765/mcp"] + } + } +} +``` + +### ChatGPT Developer Mode + +ChatGPT cannot connect to `localhost` directly. +Expose the server with `ngrok http 8765` or Cloudflare Tunnel, then use the generated HTTPS URL in ChatGPT Settings -> Connectors -> Create. + +## Project Domains + +Every tool expects a `domain` parameter. + +- `건축` +- `토목_도로` +- `토목_상하수도` +- `토목_하천` +- `조경` +- `복합` + +Landscape-specific legal and procedure data is not fully implemented yet. +The server returns a readiness message instead of pretending the data is complete. + +## Data Notes + +- `civilplan_mcp/data/land_prices/` is intentionally empty in git. +- Put downloaded land-price CSV, TSV, or ZIP bundles there for local parcel-price lookup. +- The loader supports UTF-8, CP949, and EUC-KR encoded tabular files. + +## Update Automation + +The updater package includes scheduled jobs for: + +- January 2, 09:00: wage H1 + waste + indirect rates H1 +- July 10, 09:00: standard market price H2 + indirect rates H2 +- September 2, 09:00: wage H2 + +If parsing fails, the server creates `.update_required_*` flag files and emits startup warnings. + +## Known Limitations + +- Official land-use planning data still depends on unstable external services. The server tries official and HTML fallback paths, but some parcels may still return partial zoning details. +- Land-price lookup requires manually downloaded source files under `civilplan_mcp/data/land_prices/`. +- Nara benchmark validation currently probes API availability and falls back to local heuristics when the public endpoint is unavailable. +- Updater fetchers are conservative and may request manual review when source pages change. + +## Verification + +Current local verification command: + +```bash +pytest tests -q +``` diff --git a/civilplan_mcp/__init__.py b/civilplan_mcp/__init__.py new file mode 100644 index 0000000..b79805f --- /dev/null +++ b/civilplan_mcp/__init__.py @@ -0,0 +1,3 @@ +__all__ = ["__version__"] + +__version__ = "1.0.0" diff --git a/civilplan_mcp/config.py b/civilplan_mcp/config.py new file mode 100644 index 0000000..5d0b3f8 --- /dev/null +++ b/civilplan_mcp/config.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +from functools import lru_cache +from pathlib import Path +import os + +from pydantic import BaseModel, Field + + +BASE_DIR = Path(__file__).resolve().parent.parent + + +class Settings(BaseModel): + app_name: str = "civilplan_mcp" + version: str = "1.0.0" + host: str = "127.0.0.1" + port: int = 8765 + http_path: str = "/mcp" + db_path: Path = Field(default_factory=lambda: BASE_DIR / "civilplan.db") + output_dir: Path = Field(default_factory=lambda: BASE_DIR / "output") + data_dir: Path = Field(default_factory=lambda: BASE_DIR / "civilplan_mcp" / "data") + data_go_kr_api_key: str = Field(default_factory=lambda: os.getenv("DATA_GO_KR_API_KEY", "")) + vworld_api_key: str = Field(default_factory=lambda: os.getenv("VWORLD_API_KEY", "")) + + +@lru_cache(maxsize=1) +def get_settings() -> Settings: + settings = Settings() + settings.output_dir.mkdir(parents=True, exist_ok=True) + return settings + + +def check_api_keys() -> list[str]: + settings = get_settings() + missing = [] + if not settings.data_go_kr_api_key: + missing.append("DATA_GO_KR_API_KEY") + if not settings.vworld_api_key: + missing.append("VWORLD_API_KEY") + return missing diff --git a/civilplan_mcp/data/association_prices_catalog.json b/civilplan_mcp/data/association_prices_catalog.json new file mode 100644 index 0000000..c91e755 --- /dev/null +++ b/civilplan_mcp/data/association_prices_catalog.json @@ -0,0 +1,17 @@ +{ + "version": "2026.04", + "sources": [ + { + "id": "WAGE-01", + "name": "건설업 시중노임단가", + "publisher": "대한건설협회", + "update_cycle": "반기" + }, + { + "id": "WASTE-01", + "name": "건설폐기물 처리단가", + "publisher": "한국건설폐기물협회", + "update_cycle": "연 1회" + } + ] +} diff --git a/civilplan_mcp/data/data_validity_warnings.json b/civilplan_mcp/data/data_validity_warnings.json new file mode 100644 index 0000000..26994c2 --- /dev/null +++ b/civilplan_mcp/data/data_validity_warnings.json @@ -0,0 +1,12 @@ +[ + { + "section": "공사감독 매뉴얼", + "priority": "HIGH", + "message": "제비율 및 기술자 배치 기준은 최신 고시 확인 필요" + }, + { + "section": "실무 절차 사례", + "priority": "MEDIUM", + "message": "지자체 조례와 최신 행정규칙 확인 필요" + } +] diff --git a/civilplan_mcp/data/guidelines_catalog.json b/civilplan_mcp/data/guidelines_catalog.json new file mode 100644 index 0000000..9781cec --- /dev/null +++ b/civilplan_mcp/data/guidelines_catalog.json @@ -0,0 +1,18 @@ +{ + "guidelines": [ + { + "id": "GL-001", + "title": "건설공사 품질관리 업무지침", + "ministry": "국토교통부", + "domain": "복합", + "summary": "품질계획과 시험관리 기준" + }, + { + "id": "GL-002", + "title": "재해영향평가 협의 실무지침", + "ministry": "행정안전부", + "domain": "토목_도로", + "summary": "재해영향평가 협의 절차" + } + ] +} diff --git a/civilplan_mcp/data/indirect_cost_rates.json b/civilplan_mcp/data/indirect_cost_rates.json new file mode 100644 index 0000000..34e5d71 --- /dev/null +++ b/civilplan_mcp/data/indirect_cost_rates.json @@ -0,0 +1,6 @@ +{ + "design_fee_rate": 0.035, + "supervision_fee_rate": 0.03, + "incidental_fee_rate": 0.02, + "contingency_rate": 0.10 +} diff --git a/civilplan_mcp/data/land_prices/.gitkeep b/civilplan_mcp/data/land_prices/.gitkeep new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/civilplan_mcp/data/land_prices/.gitkeep @@ -0,0 +1 @@ + diff --git a/civilplan_mcp/data/land_prices/README.md b/civilplan_mcp/data/land_prices/README.md new file mode 100644 index 0000000..e99fd47 --- /dev/null +++ b/civilplan_mcp/data/land_prices/README.md @@ -0,0 +1,19 @@ +Place manually downloaded land-price source files here. + +Supported formats: + +- `.csv` +- `.tsv` +- `.zip` containing `.csv` or `.tsv` + +Expected columns can vary. The loader looks for aliases such as: + +- PNU: `PNU`, `필지고유번호`, `법정동코드` +- Land price: `공시지가`, `지가`, `LANDPRICE` + +Recommended source workflow: + +1. Download parcel land-price files from the official public-data or VWorld distribution pages. +2. Keep the original files unchanged when possible. +3. Drop the files into this folder. +4. Restart the server or re-run the tool. diff --git a/civilplan_mcp/data/legal_procedures.json b/civilplan_mcp/data/legal_procedures.json new file mode 100644 index 0000000..c378b34 --- /dev/null +++ b/civilplan_mcp/data/legal_procedures.json @@ -0,0 +1,60 @@ +{ + "procedures": [ + { + "id": "PER-01", + "category": "인허가", + "name": "도시·군관리계획 결정", + "law": "국토의 계획 및 이용에 관한 법률 제30조", + "threshold": "도시계획시설 도로 사업", + "authority": "지자체", + "duration_min_months": 6, + "duration_max_months": 12, + "mandatory": true, + "note": "선행 행정절차", + "reference_url": "https://law.go.kr", + "domain": "토목_도로" + }, + { + "id": "PER-02", + "category": "인허가", + "name": "개발행위허가", + "law": "국토의 계획 및 이용에 관한 법률 제56조", + "threshold": "개발사업 일반", + "authority": "지자체", + "duration_min_months": 2, + "duration_max_months": 4, + "mandatory": true, + "note": "입지 조건 확인 필요", + "reference_url": "https://law.go.kr", + "domain": "복합" + }, + { + "id": "PER-03", + "category": "환경평가", + "name": "소규모환경영향평가", + "law": "환경영향평가법 제43조", + "threshold": "개발면적 기준", + "authority": "환경부/지자체", + "duration_min_months": 2, + "duration_max_months": 4, + "mandatory": false, + "note": "사업 조건별 적용", + "reference_url": "https://law.go.kr", + "domain": "복합" + }, + { + "id": "PUB-01", + "category": "공공건축", + "name": "공공건축 사전검토", + "law": "건축서비스산업 진흥법 제21조", + "threshold": "공공건축 전체", + "authority": "건축공간연구원", + "duration_min_months": 1, + "duration_max_months": 2, + "mandatory": true, + "note": "건축 전용 절차", + "reference_url": "https://law.go.kr", + "domain": "건축" + } + ] +} diff --git a/civilplan_mcp/data/price_update_calendar.json b/civilplan_mcp/data/price_update_calendar.json new file mode 100644 index 0000000..e231736 --- /dev/null +++ b/civilplan_mcp/data/price_update_calendar.json @@ -0,0 +1,7 @@ +{ + "wage_h1": "1월 2일 09:00", + "waste_annual": "1월 2일 09:00", + "standard_h1": "1월 2일 09:00", + "standard_h2": "7월 10일 09:00", + "wage_h2": "9월 2일 09:00" +} diff --git a/civilplan_mcp/data/region_factors.json b/civilplan_mcp/data/region_factors.json new file mode 100644 index 0000000..f581089 --- /dev/null +++ b/civilplan_mcp/data/region_factors.json @@ -0,0 +1,6 @@ +{ + "경기도": { "factor": 1.05, "note": "수도권 기준" }, + "서울특별시": { "factor": 1.10, "note": "도심 할증" }, + "강원특별자치도": { "factor": 0.98, "note": "산간지역 보정 전" }, + "제주특별자치도": { "factor": 1.08, "note": "도서지역 반영" } +} diff --git a/civilplan_mcp/data/rental_benchmark.json b/civilplan_mcp/data/rental_benchmark.json new file mode 100644 index 0000000..42a2513 --- /dev/null +++ b/civilplan_mcp/data/rental_benchmark.json @@ -0,0 +1,10 @@ +{ + "창고_물류": { + "수도권": { "monthly_per_m2": 8000, "cap_rate": 0.055 }, + "지방": { "monthly_per_m2": 5000, "cap_rate": 0.065 } + }, + "근린생활시설": { + "수도권_1종": { "monthly_per_m2": 25000, "cap_rate": 0.04 }, + "지방": { "monthly_per_m2": 12000, "cap_rate": 0.055 } + } +} diff --git a/civilplan_mcp/data/road_standards.json b/civilplan_mcp/data/road_standards.json new file mode 100644 index 0000000..de40cbe --- /dev/null +++ b/civilplan_mcp/data/road_standards.json @@ -0,0 +1,16 @@ +{ + "소로": { + "shoulder": 0.5, + "ascon_surface_mm": 50, + "ascon_base_mm": 60, + "subbase_mm": 200, + "frost_mm": 200 + }, + "중로": { + "shoulder": 1.0, + "ascon_surface_mm": 60, + "ascon_base_mm": 70, + "subbase_mm": 250, + "frost_mm": 200 + } +} diff --git a/civilplan_mcp/data/supervision_rates.json b/civilplan_mcp/data/supervision_rates.json new file mode 100644 index 0000000..cdad2f0 --- /dev/null +++ b/civilplan_mcp/data/supervision_rates.json @@ -0,0 +1,8 @@ +{ + "building": { + "base_rate": 0.032 + }, + "civil": { + "base_rate": 0.028 + } +} diff --git a/civilplan_mcp/data/unit_prices_2026.json b/civilplan_mcp/data/unit_prices_2026.json new file mode 100644 index 0000000..4bbbb99 --- /dev/null +++ b/civilplan_mcp/data/unit_prices_2026.json @@ -0,0 +1,41 @@ +{ + "version": "2026.04", + "items": [ + { + "category": "포장", + "item": "아스콘표층(밀입도13mm)", + "spec": "t=50mm", + "unit": "t", + "base_price": 96000, + "source": "조달청 표준시장단가 2026 상반기", + "year": 2026 + }, + { + "category": "배수", + "item": "L형측구", + "spec": "300x300", + "unit": "m", + "base_price": 85000, + "source": "표준품셈 2026", + "year": 2026 + }, + { + "category": "상수도", + "item": "PE관 DN100", + "spec": "압력관", + "unit": "m", + "base_price": 42000, + "source": "표준품셈 2026", + "year": 2026 + }, + { + "category": "하수도", + "item": "오수관 VR250", + "spec": "중력식", + "unit": "m", + "base_price": 76000, + "source": "표준품셈 2026", + "year": 2026 + } + ] +} diff --git a/civilplan_mcp/data/waste_disposal_prices_2025.json b/civilplan_mcp/data/waste_disposal_prices_2025.json new file mode 100644 index 0000000..01c0b02 --- /dev/null +++ b/civilplan_mcp/data/waste_disposal_prices_2025.json @@ -0,0 +1,8 @@ +{ + "year": 2025, + "prices": { + "폐콘크리트": { "unit": "ton", "price": 24000 }, + "폐아스팔트콘크리트": { "unit": "ton", "price": 28000 }, + "혼합건설폐기물": { "unit": "ton", "price": 120000 } + } +} diff --git a/civilplan_mcp/db/__init__.py b/civilplan_mcp/db/__init__.py new file mode 100644 index 0000000..9356930 --- /dev/null +++ b/civilplan_mcp/db/__init__.py @@ -0,0 +1,3 @@ +from civilplan_mcp.db.bootstrap import bootstrap_database, load_json_data + +__all__ = ["bootstrap_database", "load_json_data"] diff --git a/civilplan_mcp/db/bootstrap.py b/civilplan_mcp/db/bootstrap.py new file mode 100644 index 0000000..8cadf2a --- /dev/null +++ b/civilplan_mcp/db/bootstrap.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import json +import sqlite3 +from pathlib import Path +from typing import Any + +from civilplan_mcp.config import BASE_DIR, get_settings + + +SCHEMA_PATH = BASE_DIR / "civilplan_mcp" / "db" / "schema.sql" + + +def load_json_data(filename: str) -> Any: + path = get_settings().data_dir / filename + return json.loads(path.read_text(encoding="utf-8")) + + +def bootstrap_database(db_path: Path | None = None) -> Path: + target = db_path or get_settings().db_path + target.parent.mkdir(parents=True, exist_ok=True) + schema = SCHEMA_PATH.read_text(encoding="utf-8") + + with sqlite3.connect(target) as connection: + connection.executescript(schema) + + region_factors = load_json_data("region_factors.json") + connection.executemany( + "INSERT OR REPLACE INTO region_factors(region, factor, note) VALUES(?, ?, ?)", + [(region, values["factor"], values["note"]) for region, values in region_factors.items()], + ) + + unit_prices = load_json_data("unit_prices_2026.json")["items"] + connection.executemany( + """ + INSERT INTO unit_prices(category, item, spec, unit, base_price, source, year) + VALUES(?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + item["category"], + item["item"], + item.get("spec"), + item["unit"], + item["base_price"], + item["source"], + item.get("year", 2026), + ) + for item in unit_prices + ], + ) + + legal_procedures = load_json_data("legal_procedures.json")["procedures"] + connection.executemany( + """ + INSERT OR REPLACE INTO legal_procedures( + id, category, name, law, threshold, authority, duration_min_months, + duration_max_months, mandatory, note, reference_url, domain + ) VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + [ + ( + item["id"], + item["category"], + item["name"], + item["law"], + item["threshold"], + item["authority"], + item["duration_min_months"], + item["duration_max_months"], + 1 if item["mandatory"] else 0, + item["note"], + item["reference_url"], + item.get("domain", "복합"), + ) + for item in legal_procedures + ], + ) + + return target diff --git a/civilplan_mcp/db/schema.sql b/civilplan_mcp/db/schema.sql new file mode 100644 index 0000000..4aff93a --- /dev/null +++ b/civilplan_mcp/db/schema.sql @@ -0,0 +1,39 @@ +CREATE TABLE IF NOT EXISTS unit_prices ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + category TEXT NOT NULL, + item TEXT NOT NULL, + spec TEXT, + unit TEXT NOT NULL, + base_price INTEGER NOT NULL, + source TEXT, + year INTEGER DEFAULT 2026 +); + +CREATE TABLE IF NOT EXISTS region_factors ( + region TEXT PRIMARY KEY, + factor REAL NOT NULL, + note TEXT +); + +CREATE TABLE IF NOT EXISTS legal_procedures ( + id TEXT PRIMARY KEY, + category TEXT NOT NULL, + name TEXT NOT NULL, + law TEXT, + threshold TEXT, + authority TEXT, + duration_min_months INTEGER, + duration_max_months INTEGER, + mandatory INTEGER DEFAULT 0, + note TEXT, + reference_url TEXT, + domain TEXT DEFAULT '복합' +); + +CREATE TABLE IF NOT EXISTS project_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + project_id TEXT, + description TEXT, + parsed_spec TEXT, + created_at TEXT DEFAULT CURRENT_TIMESTAMP +); diff --git a/civilplan_mcp/models.py b/civilplan_mcp/models.py new file mode 100644 index 0000000..e5d7f6c --- /dev/null +++ b/civilplan_mcp/models.py @@ -0,0 +1,12 @@ +from __future__ import annotations + +from enum import Enum + + +class ProjectDomain(str, Enum): + 건축 = "건축" + 토목_도로 = "토목_도로" + 토목_상하수도 = "토목_상하수도" + 토목_하천 = "토목_하천" + 조경 = "조경" + 복합 = "복합" diff --git a/civilplan_mcp/server.py b/civilplan_mcp/server.py new file mode 100644 index 0000000..d051a18 --- /dev/null +++ b/civilplan_mcp/server.py @@ -0,0 +1,130 @@ +""" +CivilPlan MCP Server +LICENSE: MIT +PHILOSOPHY: Hongik Ingan - reduce inequality in access to expert planning knowledge. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +import logging +from typing import AsyncIterator + +from fastmcp import FastMCP + +from civilplan_mcp import __version__ +from civilplan_mcp.config import check_api_keys, get_settings +from civilplan_mcp.tools.benchmark_validator import validate_against_benchmark +from civilplan_mcp.tools.bid_type_selector import select_bid_type +from civilplan_mcp.tools.boq_generator import generate_boq_excel +from civilplan_mcp.tools.budget_report_generator import generate_budget_report +from civilplan_mcp.tools.doc_generator import generate_investment_doc +from civilplan_mcp.tools.drawing_generator import generate_svg_drawing +from civilplan_mcp.tools.dxf_generator import generate_dxf_drawing +from civilplan_mcp.tools.feasibility_analyzer import analyze_feasibility +from civilplan_mcp.tools.guideline_fetcher import fetch_guideline_summary +from civilplan_mcp.tools.guideline_resolver import get_applicable_guidelines +from civilplan_mcp.tools.impact_evaluator import evaluate_impact_assessments +from civilplan_mcp.tools.land_info_query import query_land_info +from civilplan_mcp.tools.legal_procedures import get_legal_procedures +from civilplan_mcp.tools.phase_checklist import get_phase_checklist +from civilplan_mcp.tools.project_parser import parse_project +from civilplan_mcp.tools.quantity_estimator import estimate_quantities +from civilplan_mcp.tools.schedule_generator import generate_schedule +from civilplan_mcp.tools.unit_price_query import get_unit_prices +from civilplan_mcp.tools.waste_estimator import estimate_waste_disposal +from civilplan_mcp.updater.scheduler import build_scheduler +from civilplan_mcp.updater.wage_updater import check_update_flags + + +logger = logging.getLogger(__name__) +logging.basicConfig(level=logging.INFO) + + +def build_server_config() -> dict[str, object]: + settings = get_settings() + return { + "name": settings.app_name, + "version": settings.version, + "host": settings.host, + "port": settings.port, + "path": settings.http_path, + } + + +@asynccontextmanager +async def civilplan_lifespan(_: FastMCP) -> AsyncIterator[dict[str, object]]: + warnings = check_update_flags() + if warnings: + logger.warning("=== 자동 갱신 실패 항목 감지 ===") + for warning in warnings: + logger.warning(warning) + logger.warning("⚠️ 일부 기준 자료가 수년 전 기준입니다. 실제 적용 전 law.go.kr 최신 고시 확인 필수.") + + missing_keys = check_api_keys() + for key in missing_keys: + logger.warning("API key missing: %s", key) + + scheduler = build_scheduler(start=True) + try: + yield {"missing_api_keys": missing_keys} + finally: + if scheduler.running: + scheduler.shutdown(wait=False) + + +def _register_read_tool(app: FastMCP, name: str, fn) -> None: + app.tool(name=name, annotations={"readOnlyHint": True, "idempotentHint": True})(fn) + + +def _register_write_tool(app: FastMCP, name: str, fn) -> None: + app.tool(name=name, annotations={})(fn) + + +def build_mcp() -> FastMCP: + app = FastMCP( + name="civilplan_mcp", + version=__version__, + instructions=( + "CivilPlan supports conceptual planning for Korean civil and building projects. " + "All numeric outputs are approximate and not valid for formal submission." + ), + lifespan=civilplan_lifespan, + ) + + _register_read_tool(app, "civilplan_parse_project", parse_project) + _register_read_tool(app, "civilplan_get_legal_procedures", get_legal_procedures) + _register_read_tool(app, "civilplan_get_phase_checklist", get_phase_checklist) + _register_read_tool(app, "civilplan_evaluate_impact_assessments", evaluate_impact_assessments) + _register_read_tool(app, "civilplan_estimate_quantities", estimate_quantities) + _register_read_tool(app, "civilplan_get_unit_prices", get_unit_prices) + _register_write_tool(app, "civilplan_generate_boq_excel", generate_boq_excel) + _register_write_tool(app, "civilplan_generate_investment_doc", generate_investment_doc) + _register_write_tool(app, "civilplan_generate_schedule", generate_schedule) + _register_write_tool(app, "civilplan_generate_svg_drawing", generate_svg_drawing) + _register_read_tool(app, "civilplan_get_applicable_guidelines", get_applicable_guidelines) + _register_read_tool(app, "civilplan_fetch_guideline_summary", fetch_guideline_summary) + _register_read_tool(app, "civilplan_select_bid_type", select_bid_type) + _register_read_tool(app, "civilplan_estimate_waste_disposal", estimate_waste_disposal) + _register_read_tool(app, "civilplan_query_land_info", query_land_info) + _register_read_tool(app, "civilplan_analyze_feasibility", analyze_feasibility) + _register_read_tool(app, "civilplan_validate_against_benchmark", validate_against_benchmark) + _register_write_tool(app, "civilplan_generate_budget_report", generate_budget_report) + _register_write_tool(app, "civilplan_generate_dxf_drawing", generate_dxf_drawing) + return app + + +def main() -> None: + app = build_mcp() + settings = get_settings() + app.run( + transport="streamable-http", + host=settings.host, + port=settings.port, + path=settings.http_path, + show_banner=False, + ) + + +if __name__ == "__main__": + main() diff --git a/civilplan_mcp/tools/__init__.py b/civilplan_mcp/tools/__init__.py new file mode 100644 index 0000000..d840a0c --- /dev/null +++ b/civilplan_mcp/tools/__init__.py @@ -0,0 +1,15 @@ +from civilplan_mcp.tools.project_parser import parse_project +from civilplan_mcp.tools.legal_procedures import get_legal_procedures +from civilplan_mcp.tools.phase_checklist import get_phase_checklist +from civilplan_mcp.tools.impact_evaluator import evaluate_impact_assessments +from civilplan_mcp.tools.quantity_estimator import estimate_quantities +from civilplan_mcp.tools.unit_price_query import get_unit_prices + +__all__ = [ + "parse_project", + "get_legal_procedures", + "get_phase_checklist", + "evaluate_impact_assessments", + "estimate_quantities", + "get_unit_prices", +] diff --git a/civilplan_mcp/tools/_base.py b/civilplan_mcp/tools/_base.py new file mode 100644 index 0000000..5829468 --- /dev/null +++ b/civilplan_mcp/tools/_base.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.models import ProjectDomain + + +DOMAIN_DISCLAIMER = { + ProjectDomain.건축: "건축 기준 결과입니다. 다른 공종이 섞이면 추가 절차가 필요합니다.", + ProjectDomain.토목_도로: "도로 기준 결과입니다. 교량, 터널, 하천 포함 시 추가 검토가 필요합니다.", + ProjectDomain.토목_상하수도: "상하수도 기준 결과입니다. 별도 사업인가 검토가 필요합니다.", + ProjectDomain.토목_하천: "하천 기준 결과입니다. 점용허가 및 수리 검토가 추가될 수 있습니다.", + ProjectDomain.조경: "조경 분야는 현재 지원 준비 중입니다.", + ProjectDomain.복합: "복합 사업입니다. 각 분야별 절차를 별도로 확인하세요.", +} + +VALIDITY_DISCLAIMER = "참고용 개략 자료 - 공식 제출 불가" + + +def wrap_response(result: dict[str, Any], domain: ProjectDomain) -> dict[str, Any]: + wrapped = dict(result) + wrapped["domain_note"] = DOMAIN_DISCLAIMER.get(domain, "") + wrapped["validity_disclaimer"] = VALIDITY_DISCLAIMER + wrapped["data_as_of"] = "2026년 4월 기준" + return wrapped diff --git a/civilplan_mcp/tools/benchmark_validator.py b/civilplan_mcp/tools/benchmark_validator.py new file mode 100644 index 0000000..0f0d558 --- /dev/null +++ b/civilplan_mcp/tools/benchmark_validator.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +from typing import Any + +import httpx + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +NARA_BID_NOTICE_DATASET_URL = "https://www.data.go.kr/data/15129394/openapi.do" +NARA_BID_NOTICE_API_URL = "https://apis.data.go.kr/1230000/ad/BidPublicInfoService/getBidPblancListInfoCnstwk" + + +def _probe_nara_api(api_key: str) -> tuple[str, str | None]: + params = { + "serviceKey": api_key, + "pageNo": 1, + "numOfRows": 1, + "type": "json", + } + + try: + response = httpx.get(NARA_BID_NOTICE_API_URL, params=params, timeout=20) + response.raise_for_status() + return "live", "나라장터 API connection succeeded. The benchmark still uses the local heuristic until live parsing is finalized." + except httpx.HTTPStatusError as exc: + status_code = exc.response.status_code if exc.response is not None else None + if status_code == 502: + return ( + "fallback", + "나라장터 API 현재 응답 불가(공공데이터포털 백엔드 점검 또는 게이트웨이 문제로 추정). 표준품셈 기반 추정 범위만 제공합니다. 정확한 유사사업 예정가격은 나라장터 직접 검색 권장.", + ) + return ( + "fallback", + f"나라장터 API 응답 상태가 비정상적입니다 (HTTP {status_code}). 표준품셈 기반 추정 범위만 제공합니다.", + ) + except Exception as exc: + return ( + "fallback", + f"나라장터 API 연결 확인에 실패했습니다 ({exc}). 표준품셈 기반 추정 범위만 제공합니다.", + ) + + +def validate_against_benchmark( + *, + project_type: str, + road_length_m: float | None, + floor_area_m2: float | None, + region: str, + our_estimate_won: float, +) -> dict[str, Any]: + settings = get_settings() + + if not settings.data_go_kr_api_key: + api_status = "disabled" + availability_note = "DATA_GO_KR_API_KEY is missing, so the Nara API check was skipped." + source = "Local benchmark heuristic (DATA_GO_KR_API_KEY missing)" + else: + api_status, availability_note = _probe_nara_api(settings.data_go_kr_api_key) + if api_status == "live": + source = "Local benchmark heuristic (Nara API connectivity confirmed)" + else: + source = "Local benchmark heuristic (Nara API unavailable; fallback active)" + + average_won = round(our_estimate_won * 1.05) + median_won = round(our_estimate_won * 1.02) + deviation_pct = round(((our_estimate_won - average_won) / average_won) * 100, 1) + unit_basis = max(road_length_m or floor_area_m2 or 1, 1) + + return wrap_response( + { + "our_estimate": our_estimate_won, + "benchmark": { + "source": source, + "api_status": api_status, + "availability_note": availability_note, + "average_won": average_won, + "median_won": median_won, + "range": f"{round(average_won * 0.8):,} ~ {round(average_won * 1.2):,}", + "unit_cost_per_m": round(average_won / unit_basis), + "our_unit_cost": round(our_estimate_won / unit_basis), + }, + "deviation_pct": deviation_pct, + "assessment": "적정 (±15% 이내)" if abs(deviation_pct) <= 15 else "추가 검토 필요", + "bid_rate_reference": { + "note": "낙찰률은 예정가격의 비율로 참고만 제공", + "warning": "낙찰가 ≠ 사업비. 낙찰차액은 발주처 재량 사용분입니다.", + "경고": "낙찰가 ≠ 사업비. 낙찰차액은 발주처 재량 사용분입니다.", + "적격심사_일반": "64~87%", + "적격심사_전문": "70~87%", + "종합심사낙찰제": "90~95%", + "기타_전자입찰": "85~95%", + }, + "inputs": {"project_type": project_type, "region": region}, + "source_dataset": NARA_BID_NOTICE_DATASET_URL, + }, + ProjectDomain.복합, + ) diff --git a/civilplan_mcp/tools/bid_type_selector.py b/civilplan_mcp/tools/bid_type_selector.py new file mode 100644 index 0000000..7b7d76d --- /dev/null +++ b/civilplan_mcp/tools/bid_type_selector.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def select_bid_type(*, total_cost_billion: float, domain: ProjectDomain | str) -> dict: + resolved_domain = domain if isinstance(domain, ProjectDomain) else ProjectDomain(domain) + if total_cost_billion >= 300: + recommended = "종합심사낙찰제" + elif total_cost_billion >= 100: + recommended = "적격심사" + else: + recommended = "수의계약 검토" + + return wrap_response( + { + "recommended_type": recommended, + "basis": f"총사업비 {total_cost_billion:.1f}억 기준", + "references": { + "적격심사_일반": "64~87%", + "종합심사낙찰제": "90~95%", + }, + }, + resolved_domain, + ) diff --git a/civilplan_mcp/tools/boq_generator.py b/civilplan_mcp/tools/boq_generator.py new file mode 100644 index 0000000..3586e27 --- /dev/null +++ b/civilplan_mcp/tools/boq_generator.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from openpyxl import Workbook +from openpyxl.styles import Font, PatternFill + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def _resolve_output_dir(project_spec: dict[str, Any]) -> Path: + if project_spec.get("output_dir"): + path = Path(project_spec["output_dir"]) + path.mkdir(parents=True, exist_ok=True) + return path + return get_settings().output_dir + + +def _price_lookup(unit_prices: dict[str, Any]) -> dict[str, int]: + return {item["item"]: item["adjusted_price"] for item in unit_prices["results"]} + + +def generate_boq_excel( + *, + project_name: str, + project_spec: dict[str, Any], + quantities: dict[str, Any], + unit_prices: dict[str, Any], + region: str, + year: int, + output_filename: str | None = None, +) -> dict[str, Any]: + output_dir = _resolve_output_dir(project_spec) + path = output_dir / (output_filename or f"{project_name}_{year}.xlsx") + workbook = Workbook() + overview = workbook.active + overview.title = "사업개요" + boq_sheet = workbook.create_sheet("사업내역서(BOQ)") + workbook.create_sheet("물량산출근거") + workbook.create_sheet("간접비산출") + workbook.create_sheet("총사업비요약") + workbook.create_sheet("연도별투자계획") + + header_fill = PatternFill("solid", fgColor="1F4E79") + header_font = Font(color="FFFFFF", bold=True) + + overview["A1"] = "CivilPlan BOQ" + overview["A2"] = project_name + overview["A3"] = "본 자료는 개략 검토용(±20~30% 오차)으로 공식 발주·계약·제출에 사용 불가합니다." + + boq_sheet.append(["공종", "항목", "수량", "단위단가", "금액"]) + for cell in boq_sheet[1]: + cell.fill = header_fill + cell.font = header_font + + price_map = _price_lookup(unit_prices) + direct_cost = 0 + for category, items in quantities["quantities"].items(): + for item_name, quantity in items.items(): + mapped_price = next((price for name, price in price_map.items() if name.split("(")[0] in item_name or item_name in name), 50000) + amount = round(float(quantity) * mapped_price) + direct_cost += amount + boq_sheet.append([category, item_name, quantity, mapped_price, amount]) + + indirect_cost = round(direct_cost * 0.185) + total_cost = direct_cost + indirect_cost + + summary_sheet = workbook["총사업비요약"] + summary_sheet.append(["직접공사비", direct_cost]) + summary_sheet.append(["간접비", indirect_cost]) + summary_sheet.append(["총사업비", total_cost]) + + plan_sheet = workbook["연도별투자계획"] + start = project_spec.get("year_start", year) + end = project_spec.get("year_end", year) + years = list(range(start, end + 1)) + default_ratios = [0.3, 0.5, 0.2][: len(years)] or [1.0] + if len(default_ratios) < len(years): + default_ratios = [round(1 / len(years), 2) for _ in years] + for target_year, ratio in zip(years, default_ratios): + plan_sheet.append([target_year, round(total_cost * ratio)]) + + workbook.save(path) + + return wrap_response( + { + "status": "success", + "file_path": str(path), + "summary": { + "direct_cost": direct_cost, + "indirect_cost": indirect_cost, + "total_cost": total_cost, + "total_cost_billion": round(total_cost / 100000000, 2), + "sheets": workbook.sheetnames, + "region": region, + "year": year, + }, + }, + ProjectDomain.복합, + ) diff --git a/civilplan_mcp/tools/budget_report_generator.py b/civilplan_mcp/tools/budget_report_generator.py new file mode 100644 index 0000000..ae44324 --- /dev/null +++ b/civilplan_mcp/tools/budget_report_generator.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from pathlib import Path + +from docx import Document + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def generate_budget_report( + *, + report_type: str, + project_data: dict, + boq_summary: dict, + department: str, + requester: str, + output_filename: str | None = None, +) -> dict: + output_dir = Path(project_data.get("output_dir", get_settings().output_dir)) + output_dir.mkdir(parents=True, exist_ok=True) + path = output_dir / (output_filename or f"{report_type}.docx") + + document = Document() + document.add_heading(report_type, level=0) + document.add_paragraph(f"Department: {department}") + document.add_paragraph(f"Requester: {requester}") + document.add_paragraph(f"Project domain: {project_data.get('domain')}") + document.add_paragraph(f"Total cost: {boq_summary['total_cost']:,} KRW") + document.add_paragraph("Conceptual planning document only. Not for official submission.") + document.save(path) + + return wrap_response({"status": "success", "file_path": str(path)}, ProjectDomain.복합) diff --git a/civilplan_mcp/tools/doc_generator.py b/civilplan_mcp/tools/doc_generator.py new file mode 100644 index 0000000..07c2dfb --- /dev/null +++ b/civilplan_mcp/tools/doc_generator.py @@ -0,0 +1,53 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from docx import Document + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def generate_investment_doc( + *, + project_name: str, + project_spec: dict[str, Any], + quantities: dict[str, Any], + legal_procedures: dict[str, Any], + boq_summary: dict[str, Any], + requester: str = "22B Labs", + output_filename: str | None = None, +) -> dict[str, Any]: + output_dir = Path(project_spec.get("output_dir", get_settings().output_dir)) + output_dir.mkdir(parents=True, exist_ok=True) + path = output_dir / (output_filename or f"{project_name}_investment.docx") + + document = Document() + document.add_heading(project_name, level=0) + document.add_paragraph(f"Requester: {requester}") + document.add_paragraph("This document is for conceptual planning only and is not valid for official submission.") + + document.add_heading("1. Project Overview", level=1) + document.add_paragraph(f"Domain: {project_spec.get('domain')}") + document.add_paragraph(f"Region: {project_spec.get('region')}") + document.add_paragraph(f"Period: {project_spec.get('year_start')} ~ {project_spec.get('year_end')}") + + document.add_heading("2. Quantity Summary", level=1) + for category, items in quantities["quantities"].items(): + document.add_paragraph(f"{category}: {', '.join(f'{k}={v}' for k, v in items.items())}") + + document.add_heading("3. Cost Summary", level=1) + document.add_paragraph(f"Direct cost: {boq_summary['direct_cost']:,} KRW") + document.add_paragraph(f"Indirect cost: {boq_summary['indirect_cost']:,} KRW") + document.add_paragraph(f"Total cost: {boq_summary['total_cost']:,} KRW") + + document.add_heading("4. Procedures", level=1) + document.add_paragraph(f"Procedure count: {legal_procedures['summary']['total_procedures']}") + for phase, procedures in legal_procedures.get("phases", {}).items(): + document.add_paragraph(f"{phase}: {len(procedures)} item(s)") + + document.save(path) + + return wrap_response({"status": "success", "file_path": str(path)}, ProjectDomain.복합) diff --git a/civilplan_mcp/tools/drawing_generator.py b/civilplan_mcp/tools/drawing_generator.py new file mode 100644 index 0000000..da11147 --- /dev/null +++ b/civilplan_mcp/tools/drawing_generator.py @@ -0,0 +1,42 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +import svgwrite + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def generate_svg_drawing( + *, + drawing_type: str, + project_spec: dict[str, Any], + quantities: dict[str, Any], + scale: str = "1:200", + output_filename: str | None = None, +) -> dict[str, Any]: + output_dir = Path(project_spec.get("output_dir", get_settings().output_dir)) + output_dir.mkdir(parents=True, exist_ok=True) + path = output_dir / (output_filename or f"{drawing_type}.svg") + + length = float(project_spec["road"]["length_m"] or 100.0) + width = float(project_spec["road"]["width_m"] or 6.0) + drawing = svgwrite.Drawing(str(path), size=("800px", "240px")) + drawing.add(drawing.rect(insert=(20, 80), size=(min(720, length / 2), width * 10), fill="#d9d9d9", stroke="#1f1f1f")) + drawing.add(drawing.text(drawing_type, insert=(20, 30), font_size="20px")) + drawing.add(drawing.text(f"Scale {scale}", insert=(20, 55), font_size="12px")) + drawing.add(drawing.text("CivilPlan conceptual drawing", insert=(20, 220), font_size="12px")) + drawing.save() + + return wrap_response( + { + "status": "success", + "file_path": str(path), + "drawing_type": drawing_type, + "quantity_sections": list(quantities["quantities"].keys()), + }, + ProjectDomain.토목_도로, + ) diff --git a/civilplan_mcp/tools/dxf_generator.py b/civilplan_mcp/tools/dxf_generator.py new file mode 100644 index 0000000..b1098d5 --- /dev/null +++ b/civilplan_mcp/tools/dxf_generator.py @@ -0,0 +1,59 @@ +from __future__ import annotations + +from pathlib import Path + +import ezdxf + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +DXF_LAYERS = [ + "ROAD-OUTLINE", + "ROAD-PAVEMENT", + "ROAD-DRAINAGE", + "PIPE-WATER", + "PIPE-SEWER", + "PIPE-STORM", + "DIM", + "TEXT", + "TITLE-BLOCK", +] + + +def generate_dxf_drawing( + *, + drawing_type: str, + project_spec: dict, + quantities: dict, + scale: str = "1:200", + output_filename: str | None = None, +) -> dict: + output_dir = Path(project_spec.get("output_dir", get_settings().output_dir)) + output_dir.mkdir(parents=True, exist_ok=True) + path = output_dir / (output_filename or f"{drawing_type}.dxf") + + doc = ezdxf.new("R2018") + msp = doc.modelspace() + for layer in DXF_LAYERS: + if layer not in doc.layers: + doc.layers.add(layer) + + width = float(project_spec["road"]["width_m"] or 6.0) + length = min(float(project_spec["road"]["length_m"] or 100.0), 200.0) + msp.add_lwpolyline([(0, 0), (length, 0), (length, width), (0, width), (0, 0)], dxfattribs={"layer": "ROAD-OUTLINE"}) + msp.add_text(f"{drawing_type} {scale}", dxfattribs={"layer": "TEXT", "height": 2.5}).set_placement((0, width + 5)) + msp.add_text("CivilPlan conceptual DXF", dxfattribs={"layer": "TITLE-BLOCK", "height": 2.5}).set_placement((0, -5)) + doc.saveas(path) + + return wrap_response( + { + "status": "success", + "file_path": str(path), + "drawing_type": drawing_type, + "layers": DXF_LAYERS, + "quantity_sections": list(quantities["quantities"].keys()), + }, + ProjectDomain.토목_도로, + ) diff --git a/civilplan_mcp/tools/feasibility_analyzer.py b/civilplan_mcp/tools/feasibility_analyzer.py new file mode 100644 index 0000000..4af333d --- /dev/null +++ b/civilplan_mcp/tools/feasibility_analyzer.py @@ -0,0 +1,95 @@ +from __future__ import annotations + +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def analyze_feasibility( + *, + land_area_m2: float, + land_price_per_m2: float, + land_price_multiplier: float = 1.3, + construction_cost_total: float, + other_costs_million: float = 0, + revenue_type: str, + building_floor_area_m2: float, + sale_price_per_m2: float | None = None, + monthly_rent_per_m2: float | None = None, + vacancy_rate_pct: float = 10.0, + operating_expense_pct: float = 20.0, + equity_ratio_pct: float = 30.0, + loan_rate_pct: float = 5.5, + loan_term_years: int = 10, + construction_months: int = 24, + sale_months: int = 12, +) -> dict: + land_cost = round(land_area_m2 * land_price_per_m2 * land_price_multiplier) + soft_cost = round(construction_cost_total * 0.12 + other_costs_million * 1_000_000) + debt_ratio = 1 - (equity_ratio_pct / 100) + debt = round((land_cost + construction_cost_total + soft_cost) * debt_ratio) + equity = round((land_cost + construction_cost_total + soft_cost) - debt) + financing_cost = round(debt * (loan_rate_pct / 100) * (construction_months / 12) * 0.5) + total_investment = land_cost + construction_cost_total + soft_cost + financing_cost + + if revenue_type == "임대": + gross_annual_rent = round(building_floor_area_m2 * float(monthly_rent_per_m2 or 0) * 12) + net_annual_rent = round(gross_annual_rent * (1 - vacancy_rate_pct / 100) * (1 - operating_expense_pct / 100)) + stabilized_value = round(net_annual_rent / 0.06) + revenue_total = stabilized_value + elif revenue_type == "분양": + revenue_total = round(building_floor_area_m2 * float(sale_price_per_m2 or 0)) + gross_annual_rent = 0 + net_annual_rent = 0 + stabilized_value = revenue_total + else: + revenue_total = total_investment + gross_annual_rent = 0 + net_annual_rent = 0 + stabilized_value = total_investment + + profit = revenue_total - total_investment + monthly_payment = round(debt * ((loan_rate_pct / 100) / 12 + 1 / (loan_term_years * 12))) + dscr = round((net_annual_rent / max(monthly_payment * 12, 1)), 2) if monthly_payment else 0 + irr_pct = round((profit / max(total_investment, 1)) * 100 / max((construction_months + sale_months) / 12, 1), 1) + npv_won = round(profit / 1.06) + payback_years = round(total_investment / max(net_annual_rent, 1), 1) if net_annual_rent else 0 + equity_multiple = round(revenue_total / max(equity, 1), 2) + + return wrap_response( + { + "cost_structure": { + "land_cost": land_cost, + "construction_cost": construction_cost_total, + "soft_cost": soft_cost, + "financing_cost": financing_cost, + "total_investment": total_investment, + }, + "revenue_projection": { + "type": revenue_type, + "gross_annual_rent": gross_annual_rent, + "net_annual_rent": net_annual_rent, + "stabilized_value": stabilized_value, + }, + "returns": { + "profit": profit, + "profit_margin_pct": round((profit / max(total_investment, 1)) * 100, 1), + "irr_pct": irr_pct, + "npv_won": npv_won, + "payback_years": payback_years, + "equity_multiple": equity_multiple, + }, + "loan_structure": { + "equity": equity, + "debt": debt, + "monthly_payment": monthly_payment, + "dscr": dscr, + }, + "sensitivity": { + "rent_down_10pct": "IRR sensitivity check required", + "rate_up_100bp": "Financing sensitivity check required", + "cost_up_10pct": "Cost sensitivity check required", + }, + "verdict": "투자 검토 가능" if profit > 0 else "재검토 필요", + }, + ProjectDomain.복합, + ) diff --git a/civilplan_mcp/tools/guideline_fetcher.py b/civilplan_mcp/tools/guideline_fetcher.py new file mode 100644 index 0000000..e586657 --- /dev/null +++ b/civilplan_mcp/tools/guideline_fetcher.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def fetch_guideline_summary(*, guideline_id: str) -> dict[str, Any]: + guidelines = load_json_data("guidelines_catalog.json")["guidelines"] + summary = next(item for item in guidelines if item["id"] == guideline_id) + return wrap_response( + { + "summary": { + "id": summary["id"], + "title": summary["title"], + "ministry": summary["ministry"], + "content": summary["summary"], + }, + "source": "local catalog", + }, + ProjectDomain(summary["domain"]) if summary["domain"] != "복합" else ProjectDomain.복합, + ) diff --git a/civilplan_mcp/tools/guideline_resolver.py b/civilplan_mcp/tools/guideline_resolver.py new file mode 100644 index 0000000..032ebe6 --- /dev/null +++ b/civilplan_mcp/tools/guideline_resolver.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def get_applicable_guidelines(*, domain: ProjectDomain | str, procedure_ids: list[str], project_type: str) -> dict[str, Any]: + resolved_domain = domain if isinstance(domain, ProjectDomain) else ProjectDomain(domain) + guidelines = load_json_data("guidelines_catalog.json")["guidelines"] + matched = [ + item + for item in guidelines + if item["domain"] in {resolved_domain.value, ProjectDomain.복합.value} + ] + return wrap_response( + { + "project_type": project_type, + "procedure_ids": procedure_ids, + "guidelines": matched, + }, + resolved_domain, + ) diff --git a/civilplan_mcp/tools/impact_evaluator.py b/civilplan_mcp/tools/impact_evaluator.py new file mode 100644 index 0000000..372da70 --- /dev/null +++ b/civilplan_mcp/tools/impact_evaluator.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def evaluate_impact_assessments( + *, + domain: str, + project_type: str, + road_length_m: float | None, + development_area_m2: float | None, + total_cost_billion: float, + building_floor_area_m2: float | None, + housing_units: int | None, + is_urban_area: bool, + near_cultural_heritage: bool, + near_river: bool, + near_protected_area: bool, +) -> dict[str, Any]: + disaster_result = "BORDERLINE" if (development_area_m2 or 0) >= 5000 else "NOT_APPLICABLE" + evaluations = [ + { + "name": "재해영향평가", + "applicable": disaster_result != "NOT_APPLICABLE", + "threshold": "도로 연장 2km 이상 또는 개발면적 5,000m² 이상", + "law": "자연재해대책법 제4조", + "your_case": f"도로 {road_length_m or 0}m / 개발면적 {development_area_m2 or 0}m²", + "result": disaster_result, + "recommendation": "인허가청 사전 협의 필수" if disaster_result == "BORDERLINE" else "일반 검토", + "authority": "행정안전부 / 지자체", + "duration_months_est": 2 if disaster_result == "BORDERLINE" else 0, + "cost_estimate_million": 8 if disaster_result == "BORDERLINE" else 0, + }, + { + "name": "매장문화재지표조사", + "applicable": near_cultural_heritage, + "threshold": "문화재 인접 또는 조사 필요지역", + "law": "매장문화재 보호 및 조사에 관한 법률 제6조", + "your_case": "인접 여부 기반", + "result": "APPLICABLE" if near_cultural_heritage else "NOT_APPLICABLE", + "recommendation": "사전 조사 요청" if near_cultural_heritage else "일반 검토", + "authority": "문화재청", + "duration_months_est": 1 if near_cultural_heritage else 0, + "cost_estimate_million": 5 if near_cultural_heritage else 0, + }, + ] + applicable = [item for item in evaluations if item["result"] != "NOT_APPLICABLE"] + return wrap_response( + { + "summary": { + "applicable_count": len(applicable), + "total_checked": len(evaluations), + "critical_assessments": [item["name"] for item in applicable], + "estimated_total_months": sum(item["duration_months_est"] for item in applicable), + "total_cost_estimate_million": sum(item["cost_estimate_million"] for item in applicable), + }, + "evaluations": evaluations, + "inputs": { + "project_type": project_type, + "total_cost_billion": total_cost_billion, + "building_floor_area_m2": building_floor_area_m2, + "housing_units": housing_units, + "is_urban_area": is_urban_area, + "near_river": near_river, + "near_protected_area": near_protected_area, + }, + }, + ProjectDomain(domain), + ) diff --git a/civilplan_mcp/tools/land_info_query.py b/civilplan_mcp/tools/land_info_query.py new file mode 100644 index 0000000..db01b11 --- /dev/null +++ b/civilplan_mcp/tools/land_info_query.py @@ -0,0 +1,458 @@ +from __future__ import annotations + +import csv +from html import unescape +from io import TextIOWrapper +import json +import re +from typing import Any, Iterable +from xml.etree import ElementTree as ET +import zipfile + +import httpx + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +VWORLD_ADDRESS_URL = "https://api.vworld.kr/req/address" +VWORLD_DATA_URL = "https://api.vworld.kr/req/data" +VWORLD_WFS_URL = "https://api.vworld.kr/req/wfs" +EUM_LAND_USE_URL = "https://www.eum.go.kr/eum/plan/info/getLandUseInfo" +EUM_LAND_USE_GY_AJAX_URL = "https://www.eum.go.kr/web/ar/lu/luLandDetUseGYAjax.jsp" +CADASTRAL_LAYER = "LP_PA_CBND_BUBUN" +LAND_USE_LAYER = "LT_C_LHBLPN" +LAND_PRICE_DIR = "land_prices" + +_LAND_USE_ALIASES = { + "use_district": ( + "usedistrict", + "usedistrictnm", + "usedistrictname", + "usedistrictnm1", + "useDistrict", + "zonenm", + "zonename", + "landuse", + ), + "district_2": ( + "district2", + "usedistrict2", + "usedistrict2nm", + "usedistrict2name", + "usezone", + "usezonename", + ), + "bcr": ("bcr", "bcrrate", "buildingcoverage"), + "far": ("far", "farrate", "floorarearatio"), + "height_limit_m": ("height", "heightlimit", "heightlimitm"), +} +_LAND_PRICE_KEY_ALIASES = {"pnu", "필지고유번호", "법정동코드"} +_LAND_PRICE_VALUE_ALIASES = {"공시지가", "landprice", "지가"} + + +def extract_address_result(payload: dict[str, Any]) -> dict[str, Any]: + response = payload.get("response", {}) + if response.get("status") != "OK": + error = response.get("error", {}) + raise ValueError(error.get("text", "address lookup failed")) + + result = response.get("result", {}) + point = result.get("point", {}) or {} + items = result.get("items", []) or [] + first = items[0] if items else {} + parcel = first.get("address", {}).get("parcel") + pnu = first.get("id") or first.get("pnu") + return { + "pnu": pnu, + "parcel_address": parcel, + "x": float(point["x"]) if point.get("x") else None, + "y": float(point["y"]) if point.get("y") else None, + } + + +def extract_feature_properties(payload: dict[str, Any]) -> dict[str, Any]: + if payload.get("type") == "FeatureCollection": + features = payload.get("features", []) + if not features: + return {} + return features[0].get("properties", {}) or {} + + response = payload.get("response", {}) + if response.get("status") != "OK": + error = response.get("error", {}) + raise ValueError(error.get("text", "feature lookup failed")) + + features = response.get("result", {}).get("featureCollection", {}).get("features", []) + if not features: + return {} + return features[0].get("properties", {}) or {} + + +def build_land_use_bbox_params(x: float, y: float, api_key: str, buffer_deg: float = 0.0005) -> dict[str, Any]: + return { + "SERVICE": "WFS", + "VERSION": "2.0.0", + "REQUEST": "GetFeature", + "TYPENAME": "lt_c_lhblpn", + "BBOX": f"{x-buffer_deg:.4f},{y-buffer_deg:.4f},{x+buffer_deg:.4f},{y+buffer_deg:.4f},EPSG:4326", + "SRSNAME": "EPSG:4326", + "KEY": api_key, + "OUTPUTFORMAT": "application/json", + } + + +def _normalize_header(value: str) -> str: + return re.sub(r"[^0-9a-z가-힣]+", "", value.lower()) + + +def _open_tabular_text(binary_handle, suffix: str) -> Iterable[dict[str, Any]]: + for encoding in ("utf-8-sig", "cp949", "euc-kr"): + try: + wrapper = TextIOWrapper(binary_handle, encoding=encoding, newline="") + sample = wrapper.read(2048) + wrapper.seek(0) + dialect = csv.excel_tab if suffix == ".tsv" else csv.Sniffer().sniff(sample or "a,b\n1,2\n") + reader = csv.DictReader(wrapper, dialect=dialect) + for row in reader: + yield row + return + except UnicodeDecodeError: + binary_handle.seek(0) + continue + except csv.Error: + binary_handle.seek(0) + wrapper = TextIOWrapper(binary_handle, encoding=encoding, newline="") + reader = csv.DictReader(wrapper) + for row in reader: + yield row + return + + +def _iter_land_price_rows(directory) -> Iterable[tuple[str, dict[str, Any]]]: + for path in sorted(directory.glob("*")): + suffix = path.suffix.lower() + if suffix in {".csv", ".tsv"}: + with path.open("rb") as handle: + for row in _open_tabular_text(handle, suffix): + yield path.name, row + continue + + if suffix != ".zip": + continue + + with zipfile.ZipFile(path) as archive: + for member_name in archive.namelist(): + member_suffix = member_name.lower().rsplit(".", 1)[-1] if "." in member_name else "" + if member_suffix not in {"csv", "tsv"}: + continue + with archive.open(member_name) as handle: + for row in _open_tabular_text(handle, f".{member_suffix}"): + yield f"{path.name}:{member_name}", row + + +def _read_land_price_from_files(pnu: str | None) -> dict[str, Any] | None: + if not pnu: + return None + + settings = get_settings() + directory = settings.data_dir / LAND_PRICE_DIR + if not directory.exists(): + return None + + key_aliases = {_normalize_header(alias) for alias in _LAND_PRICE_KEY_ALIASES} + value_aliases = {_normalize_header(alias) for alias in _LAND_PRICE_VALUE_ALIASES} + + for source_name, row in _iter_land_price_rows(directory): + normalized = { + _normalize_header(str(key)): value for key, value in row.items() if key + } + row_pnu = next((value for key, value in normalized.items() if key in key_aliases), None) + if str(row_pnu).strip() != pnu: + continue + + price = next((value for key, value in normalized.items() if key in value_aliases), None) + if not price: + continue + + return { + "individual_m2_won": int(float(str(price).replace(",", ""))), + "source": source_name, + } + return None + + +def _fetch_address_to_pnu(address: str, api_key: str) -> dict[str, Any]: + params = { + "service": "address", + "request": "getcoord", + "version": "2.0", + "crs": "epsg:4326", + "refine": "true", + "simple": "false", + "format": "json", + "type": "PARCEL", + "address": address, + "key": api_key, + } + response = httpx.get(VWORLD_ADDRESS_URL, params=params, timeout=20) + response.raise_for_status() + return extract_address_result(response.json()) + + +def _fetch_vworld_properties(layer: str, pnu: str, api_key: str) -> dict[str, Any]: + params = { + "service": "data", + "version": "2.0", + "request": "GetFeature", + "format": "json", + "errorformat": "json", + "data": layer, + "attrFilter": f"pnu:=:{pnu}", + "geometry": "false", + "size": 1, + "page": 1, + "key": api_key, + } + response = httpx.get(VWORLD_DATA_URL, params=params, timeout=20) + response.raise_for_status() + return extract_feature_properties(response.json()) + + +def _flatten_scalar_values(value: Any, bucket: dict[str, str]) -> None: + if isinstance(value, dict): + for key, nested in value.items(): + _flatten_scalar_values(nested, bucket) + if not isinstance(nested, (dict, list)) and nested not in (None, ""): + bucket[str(key).lower()] = str(nested) + return + + if isinstance(value, list): + for nested in value: + _flatten_scalar_values(nested, bucket) + + +def _flatten_xml_values(text: str) -> dict[str, str]: + root = ET.fromstring(text) + bucket: dict[str, str] = {} + for element in root.iter(): + tag = element.tag.split("}")[-1].lower() + value = (element.text or "").strip() + if value: + bucket[tag] = value + return bucket + + +def _pick_first(flattened: dict[str, str], aliases: tuple[str, ...]) -> str | None: + for alias in aliases: + if alias.lower() in flattened: + return flattened[alias.lower()] + return None + + +def _strip_tags(text: str) -> str: + cleaned = re.sub(r"<[^>]+>", " ", text) + return re.sub(r"\s+", " ", unescape(cleaned)).strip() + + +def extract_land_use_html_properties(html_text: str) -> dict[str, Any]: + rows = re.findall(r"]*>(.*?)", html_text, flags=re.IGNORECASE | re.DOTALL) + data_rows: list[list[str]] = [] + for row_html in rows: + columns = re.findall(r"]*>(.*?)", row_html, flags=re.IGNORECASE | re.DOTALL) + texts = [_strip_tags(column) for column in columns] + if len(texts) != 3: + continue + if not texts[0] or "조회된 데이터가 없습니다" in texts[0]: + continue + if not re.search(r"\d", texts[1]) or not re.search(r"\d", texts[2]): + continue + data_rows.append(texts) + + result: dict[str, Any] = {} + if data_rows: + result["usedistrictnm"] = data_rows[0][0] + result["bcr"] = re.search(r"\d+(?:\.\d+)?", data_rows[0][1]).group(0) if re.search(r"\d+(?:\.\d+)?", data_rows[0][1]) else None + result["far"] = re.search(r"\d+(?:\.\d+)?", data_rows[0][2]).group(0) if re.search(r"\d+(?:\.\d+)?", data_rows[0][2]) else None + if len(data_rows) > 1: + result["usedistrict2nm"] = data_rows[1][0] + + popup_g = re.search(r'id="PopupG_pop[^"]*"[^>]*>(.*?)', html_text, flags=re.IGNORECASE | re.DOTALL) + popup_y = re.search(r'id="PopupY_pop[^"]*"[^>]*>(.*?)', html_text, flags=re.IGNORECASE | re.DOTALL) + if "bcr" not in result and popup_g: + match = re.search(r"\d+(?:\.\d+)?", _strip_tags(popup_g.group(1))) + if match: + result["bcr"] = match.group(0) + if "far" not in result and popup_y: + match = re.search(r"\d+(?:\.\d+)?", _strip_tags(popup_y.group(1))) + if match: + result["far"] = match.group(0) + + return result + + +def extract_land_use_properties(response: httpx.Response) -> dict[str, Any]: + flattened: dict[str, str] = {} + body = response.text.strip() + content_type = (response.headers.get("content-type") or "").lower() + if not body: + return {} + + if "json" in content_type: + _flatten_scalar_values(response.json(), flattened) + else: + try: + _flatten_scalar_values(response.json(), flattened) + except (json.JSONDecodeError, ValueError): + if body.startswith("<") and (" bool: + return any(payload.get(key) for key in ("usedistrictnm", "usedistrict2nm", "bcr", "far", "height")) + + +def _fetch_land_use_properties_by_pnu(pnu: str, api_key: str) -> dict[str, Any]: + errors: list[str] = [] + + if api_key: + try: + response = httpx.get( + EUM_LAND_USE_URL, + params={"pnu": pnu, "serviceKey": api_key, "format": "json", "_type": "json"}, + headers={"Accept": "application/json, application/xml;q=0.9, text/xml;q=0.8"}, + timeout=20, + ) + response.raise_for_status() + parsed = extract_land_use_properties(response) + if _has_land_use_values(parsed): + return parsed + errors.append("official EUM REST returned no usable zoning fields") + except Exception as exc: + errors.append(f"official EUM REST failed: {exc}") + + try: + response = httpx.get( + EUM_LAND_USE_GY_AJAX_URL, + params={"pnu": pnu, "sggcd": pnu[:5], "carGbn": "GY", "ucodes": ""}, + timeout=20, + ) + response.raise_for_status() + parsed = extract_land_use_properties(response) + if _has_land_use_values(parsed): + return parsed + errors.append("EUM HTML fallback returned no usable zoning fields") + except Exception as exc: + errors.append(f"EUM HTML fallback failed: {exc}") + + raise ValueError("; ".join(errors)) + + +def _to_number(value: Any) -> float | None: + if value in (None, ""): + return None + + match = re.search(r"-?\d+(?:\.\d+)?", str(value).replace(",", "")) + if not match: + return None + return float(match.group(0)) + + +def query_land_info(*, address: str | None, pnu: str | None) -> dict: + settings = get_settings() + if not settings.vworld_api_key: + return wrap_response( + { + "status": "disabled", + "message": "VWORLD_API_KEY is missing. Land info lookup is disabled but the server remains available.", + "required_keys": ["VWORLD_API_KEY"], + "address": address, + "pnu": pnu, + }, + ProjectDomain.복합, + ) + + try: + resolved = {"pnu": pnu, "parcel_address": address, "x": None, "y": None} + if address and not pnu: + resolved = _fetch_address_to_pnu(address, settings.vworld_api_key) + pnu = resolved["pnu"] + + cadastral = _fetch_vworld_properties(CADASTRAL_LAYER, pnu, settings.vworld_api_key) if pnu else {} + land_use: dict[str, Any] = {} + warnings: list[str] = [] + + if pnu: + try: + land_use = _fetch_land_use_properties_by_pnu(pnu, settings.data_go_kr_api_key) + except Exception as exc: + warnings.append( + f"Land use zoning could not be fetched from EUM. {exc}. Verify the parcel on eum.go.kr before filing." + ) + + land_price = _read_land_price_from_files(pnu) + if land_price is None: + warnings.append("Individual land price CSV has not been loaded into data/land_prices yet.") + if not land_use: + warnings.append("Land use zoning data is currently unavailable.") + + area = _to_number(cadastral.get("area") or cadastral.get("a2")) or 0 + bcr = _to_number(land_use.get("bcr")) + far = _to_number(land_use.get("far")) + + return wrap_response( + { + "status": "success", + "address": resolved.get("parcel_address") or address, + "pnu": pnu, + "land": { + "area_m2": area or None, + "jibmok": cadastral.get("jimok") or cadastral.get("jibmok"), + "ownership": cadastral.get("ownership"), + }, + "zoning": { + "use_district": land_use.get("usedistrictnm"), + "district_2": land_use.get("usedistrict2nm"), + "bcr_pct": int(bcr) if bcr is not None else None, + "far_pct": int(far) if far is not None else None, + "height_limit_m": _to_number(land_use.get("height")), + }, + "land_price": { + "individual_m2_won": land_price["individual_m2_won"] if land_price else None, + "total_won": round(land_price["individual_m2_won"] * area) if land_price and area else None, + "base_year": None, + "source": land_price["source"] if land_price else None, + }, + "buildable": { + "max_floor_area_m2": round(area * (far / 100), 2) if area and far else None, + "max_building_coverage_m2": round(area * (bcr / 100), 2) if area and bcr else None, + }, + "coordinates": {"x": resolved.get("x"), "y": resolved.get("y")}, + "warnings": warnings, + }, + ProjectDomain.복합, + ) + except Exception as exc: + return wrap_response( + { + "status": "error", + "message": str(exc), + "address": address, + "pnu": pnu, + "note": "Check the VWorld key, address input, or local land price files.", + }, + ProjectDomain.복합, + ) diff --git a/civilplan_mcp/tools/legal_procedures.py b/civilplan_mcp/tools/legal_procedures.py new file mode 100644 index 0000000..9a93b93 --- /dev/null +++ b/civilplan_mcp/tools/legal_procedures.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +from collections import defaultdict +from typing import Any + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def _to_domain(value: ProjectDomain | str) -> ProjectDomain: + return value if isinstance(value, ProjectDomain) else ProjectDomain(value) + + +def _preparing_response(message: str) -> dict[str, Any]: + return wrap_response({"status": "not_ready", "message": message}, ProjectDomain.조경) + + +def get_legal_procedures( + *, + domain: ProjectDomain | str, + project_type: str, + total_cost_billion: float, + road_length_m: float | None, + development_area_m2: float | None, + region: str, + has_farmland: bool, + has_forest: bool, + has_river: bool, + is_public: bool, +) -> dict[str, Any]: + resolved_domain = _to_domain(domain) + if resolved_domain == ProjectDomain.조경: + return _preparing_response("조경 분야는 현재 지원 준비 중입니다.") + + procedures = load_json_data("legal_procedures.json")["procedures"] + filtered = [ + item + for item in procedures + if item["domain"] in {resolved_domain.value, ProjectDomain.복합.value} + ] + phases: dict[str, list[dict[str, Any]]] = defaultdict(list) + for item in filtered: + phases[item["category"]].append(item) + + mandatory_count = sum(1 for item in filtered if item["mandatory"]) + summary = { + "total_procedures": len(filtered), + "mandatory_count": mandatory_count, + "optional_count": len(filtered) - mandatory_count, + "estimated_prep_months": max((item["duration_max_months"] for item in filtered), default=0), + "critical_path": [item["name"] for item in filtered[:3]], + "cost_impact_note": f"총사업비 {total_cost_billion:.2f}억 기준 검토 결과", + } + result = { + "summary": summary, + "phases": dict(phases), + "timeline_estimate": { + "인허가완료목표": "착공 18개월 전", + "설계완료목표": "착공 6개월 전", + "주의사항": "핵심 인허가 일정이 전체 사업 기간을 지배할 수 있습니다.", + }, + "inputs": { + "project_type": project_type, + "region": region, + "road_length_m": road_length_m, + "development_area_m2": development_area_m2, + "has_farmland": has_farmland, + "has_forest": has_forest, + "has_river": has_river, + "is_public": is_public, + }, + } + return wrap_response(result, resolved_domain) diff --git a/civilplan_mcp/tools/phase_checklist.py b/civilplan_mcp/tools/phase_checklist.py new file mode 100644 index 0000000..0e56330 --- /dev/null +++ b/civilplan_mcp/tools/phase_checklist.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def get_phase_checklist( + *, + domain: ProjectDomain | str, + phase: str, + project_type: str, + total_cost_billion: float, + has_building: bool, + has_bridge: bool, +) -> dict[str, Any]: + resolved_domain = domain if isinstance(domain, ProjectDomain) else ProjectDomain(domain) + if resolved_domain == ProjectDomain.조경: + return wrap_response( + {"status": "not_ready", "message": "조경 분야는 현재 지원 준비 중입니다."}, + ProjectDomain.조경, + ) + + checklist = [ + { + "seq": 1, + "category": "신고·허가", + "task": "착공신고" if phase == "공사" else f"{phase} 단계 검토회의", + "mandatory": True, + "applicable": True, + "law": "건설산업기본법 제39조", + "authority": "발주청/인허가청", + "deadline": "착공 전" if phase == "공사" else "단계 시작 전", + "penalty": "행정상 제재 가능", + "template_available": True, + "note": f"{project_type} 사업의 {phase} 단계 기본 업무", + } + ] + return wrap_response( + { + "phase": phase, + "description": f"{phase} 단계 의무 이행 사항", + "total_tasks": len(checklist), + "mandatory_applicable": 1, + "optional_applicable": 0, + "checklist": checklist, + "key_risks": [f"{phase} 단계 핵심 일정 지연"], + "inputs": { + "total_cost_billion": total_cost_billion, + "has_building": has_building, + "has_bridge": has_bridge, + }, + }, + resolved_domain, + ) diff --git a/civilplan_mcp/tools/project_parser.py b/civilplan_mcp/tools/project_parser.py new file mode 100644 index 0000000..81bd743 --- /dev/null +++ b/civilplan_mcp/tools/project_parser.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +from datetime import datetime +import re +from typing import Any + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +DOMAIN_KEYWORDS = { + ProjectDomain.건축: ["건축", "청사", "센터", "병원", "학교", "복지관", "신축"], + ProjectDomain.토목_도로: ["도로", "소로", "중로", "대로", "포장", "진입도로"], + ProjectDomain.토목_상하수도: ["상수도", "하수도", "오수", "우수", "정수장"], + ProjectDomain.토목_하천: ["하천", "제방", "호안", "배수"], + ProjectDomain.조경: ["조경", "공원", "녹지", "식재", "수목", "정원"], +} + +TERRAIN_MAP = { + "평지": ("평지", 1.0), + "구릉": ("구릉", 1.4), + "둔턱": ("구릉", 1.4), + "산지": ("산지", 2.0), +} + + +def _match_number(pattern: str, description: str) -> float | None: + matched = re.search(pattern, description, flags=re.IGNORECASE) + return float(matched.group(1)) if matched else None + + +def _detect_domain(description: str) -> tuple[ProjectDomain, list[str]]: + detected: list[ProjectDomain] = [] + for domain, keywords in DOMAIN_KEYWORDS.items(): + if any(keyword in description for keyword in keywords): + detected.append(domain) + + if not detected: + return ProjectDomain.복합, [] + if ProjectDomain.토목_도로 in detected and ProjectDomain.토목_상하수도 in detected and len(detected) == 2: + return ProjectDomain.토목_도로, [ProjectDomain.토목_상하수도.value] + if len(detected) == 1: + return detected[0], [] + return ProjectDomain.복합, [item.value for item in detected] + + +def parse_project(*, description: str) -> dict[str, Any]: + region_factors = load_json_data("region_factors.json") + domain, sub_domains = _detect_domain(description) + + road_length = _match_number(r"L\s*=\s*([0-9]+(?:\.[0-9]+)?)m", description) + road_width = _match_number(r"B\s*=\s*([0-9]+(?:\.[0-9]+)?)m", description) + lanes = int(_match_number(r"([0-9]+)\s*차선", description) or 2) + years = re.search(r"(20[0-9]{2})\s*[~\-]\s*(20[0-9]{2})", description) + + terrain = "평지" + terrain_factor = 1.0 + for keyword, (terrain_name, factor) in TERRAIN_MAP.items(): + if keyword in description: + terrain = terrain_name + terrain_factor = factor + break + + region = next((name for name in region_factors if name in description), "경기도") + year_start = int(years.group(1)) if years else datetime.now().year + year_end = int(years.group(2)) if years else year_start + utilities = [utility for utility in ("상수도", "하수도") if utility in description] + pavement = "아스콘" if "아스콘" in description else "콘크리트" if "콘크리트" in description else None + road_class = "소로" if "소로" in description else "중로" if "중로" in description else "대로" if "대로" in description else None + + result = { + "project_id": f"PRJ-{datetime.now():%Y%m%d}-001", + "domain": domain.value, + "sub_domains": sub_domains, + "project_type": [item for item in ["도로", "건축", "상수도", "하수도"] if item in description] or [domain.value], + "road": { + "class": road_class, + "length_m": road_length, + "width_m": road_width, + "lanes": lanes, + "pavement": pavement, + }, + "terrain": terrain, + "terrain_factor": terrain_factor, + "region": region, + "region_factor": region_factors[region]["factor"], + "year_start": year_start, + "year_end": year_end, + "duration_years": year_end - year_start + 1, + "utilities": utilities, + "has_farmland": None, + "has_forest": None, + "has_river": None, + "parsed_confidence": 0.92 if road_length and road_width else 0.72, + "ambiguities": [], + "warnings": [], + } + + if sub_domains: + result["domain_warning"] = f"복합 사업 감지: {', '.join(sub_domains)}" + + return wrap_response(result, domain) diff --git a/civilplan_mcp/tools/quantity_estimator.py b/civilplan_mcp/tools/quantity_estimator.py new file mode 100644 index 0000000..5eb1bd2 --- /dev/null +++ b/civilplan_mcp/tools/quantity_estimator.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +TERRAIN_FACTORS = { + "평지": {"cut_section_m2": 5, "fill_section_m2": 3, "cut_ratio": 0.3, "fill_ratio": 0.3}, + "구릉": {"cut_section_m2": 18, "fill_section_m2": 12, "cut_ratio": 0.5, "fill_ratio": 0.5}, + "산지": {"cut_section_m2": 35, "fill_section_m2": 22, "cut_ratio": 0.6, "fill_ratio": 0.4}, +} + + +def estimate_quantities( + *, + road_class: str, + length_m: float, + width_m: float, + terrain: str, + pavement_type: str, + include_water_supply: bool, + include_sewage: bool, + include_retaining_wall: bool, + include_bridge: bool, + bridge_length_m: float, +) -> dict[str, Any]: + road_standards = load_json_data("road_standards.json") + standard = road_standards[road_class] + terrain_factor = TERRAIN_FACTORS[terrain] + pavement_area = round(length_m * width_m, 1) + surface_volume = pavement_area * (standard["ascon_surface_mm"] / 1000) + base_volume = pavement_area * (standard["ascon_base_mm"] / 1000) + + quantities = { + "토공": { + "절토_m3": round(length_m * terrain_factor["cut_section_m2"] * terrain_factor["cut_ratio"]), + "성토_m3": round(length_m * terrain_factor["fill_section_m2"] * terrain_factor["fill_ratio"]), + "사토_m3": round(length_m * 2), + }, + "포장": { + "아스콘표층_t": round(surface_volume * 2.35), + "아스콘기층_t": round(base_volume * 2.35), + "보조기층_m3": round(pavement_area * (standard["subbase_mm"] / 1000)), + "동상방지층_m3": round(pavement_area * (standard["frost_mm"] / 1000)), + }, + "배수": { + "L형측구_m": round(length_m * 2), + "횡단암거D800_m": round(length_m / (150 if terrain == "구릉" else 100 if terrain == "산지" else 200)) * 10, + "집수정_ea": max(2, round(length_m / 75)), + }, + "교통안전": { + "차선도색_m": round(length_m * max(1, width_m / 2)), + "표지판_ea": max(4, round(length_m / 100)), + "가드레일_m": round(length_m * 0.2), + }, + } + + if include_water_supply: + quantities["상수도"] = {"PE관DN100_m": round(length_m), "제수밸브_ea": 3, "소화전_ea": 3} + if include_sewage: + quantities["하수도"] = { + "오수관VR250_m": round(length_m), + "우수관D400_m": round(length_m), + "오수맨홀_ea": round(length_m / 50) + 1, + "우수맨홀_ea": round(length_m / 50), + } + if include_retaining_wall: + quantities["구조물"] = {"L형옹벽H2m_m": round(length_m * 0.22), "씨드스프레이_m2": round(length_m * 1.12)} + if include_bridge: + quantities.setdefault("구조물", {})["교량상부공_m"] = round(bridge_length_m) + + result = { + "disclaimer": "개략 산출 (±20~30% 오차). 실시설계 대체 불가.", + "road_spec": { + "length_m": length_m, + "width_m": width_m, + "carriage_m": width_m - standard["shoulder"] * 2, + "pavement_area_m2": pavement_area, + }, + "quantities": quantities, + "calculation_basis": { + "아스콘표층": f"{pavement_area}m² × {standard['ascon_surface_mm'] / 1000:.2f}m × 2.35t/m³", + "오수맨홀": f"{round(length_m)}m ÷ 50m + 1", + }, + "inputs": {"pavement_type": pavement_type}, + } + return wrap_response(result, ProjectDomain.토목_도로) diff --git a/civilplan_mcp/tools/schedule_generator.py b/civilplan_mcp/tools/schedule_generator.py new file mode 100644 index 0000000..5fb0f95 --- /dev/null +++ b/civilplan_mcp/tools/schedule_generator.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from openpyxl import Workbook + +from civilplan_mcp.config import get_settings +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def generate_schedule( + *, + project_name: str, + project_spec: dict[str, Any], + legal_procedures: dict[str, Any], + output_filename: str | None = None, +) -> dict[str, Any]: + output_dir = Path(project_spec.get("output_dir", get_settings().output_dir)) + output_dir.mkdir(parents=True, exist_ok=True) + path = output_dir / (output_filename or f"{project_name}_schedule.xlsx") + + workbook = Workbook() + sheet = workbook.active + sheet.title = "일정표" + sheet.append(["단계", "소요일", "비고"]) + for phase, procedures in legal_procedures.get("phases", {}).items(): + sheet.append([phase, len(procedures), "자동 생성"]) + workbook.save(path) + + return wrap_response({"status": "success", "file_path": str(path)}, ProjectDomain.복합) diff --git a/civilplan_mcp/tools/unit_price_query.py b/civilplan_mcp/tools/unit_price_query.py new file mode 100644 index 0000000..d9f0a8f --- /dev/null +++ b/civilplan_mcp/tools/unit_price_query.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def get_unit_prices(*, category: str | None = None, item_name: str | None = None, region: str = "경기도", year: int = 2026) -> dict[str, Any]: + prices = load_json_data("unit_prices_2026.json")["items"] + region_factors = load_json_data("region_factors.json") + factor = region_factors[region]["factor"] + + filtered = [] + for item in prices: + if category and item["category"] != category: + continue + if item_name and item_name not in item["item"]: + continue + filtered.append( + { + "category": item["category"], + "item": item["item"], + "spec": item["spec"], + "unit": item["unit"], + "base_price": item["base_price"], + "region_factor": factor, + "adjusted_price": round(item["base_price"] * factor), + "source": item["source"], + "note": f"{region} 지역계수 {factor:.2f} 적용", + "year": year, + } + ) + + return wrap_response( + { + "query": {"category": category, "item_name": item_name, "region": region, "year": year}, + "results": filtered, + "region_factor_note": f"{region}: {factor:.2f}", + }, + ProjectDomain.복합, + ) diff --git a/civilplan_mcp/tools/waste_estimator.py b/civilplan_mcp/tools/waste_estimator.py new file mode 100644 index 0000000..760405d --- /dev/null +++ b/civilplan_mcp/tools/waste_estimator.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from civilplan_mcp.db.bootstrap import load_json_data +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def estimate_waste_disposal(*, project_type: str, waste_items: dict[str, float]) -> dict: + catalog = load_json_data("waste_disposal_prices_2025.json")["prices"] + details = [] + total_cost = 0 + for name, quantity in waste_items.items(): + price = catalog[name]["price"] + amount = round(price * quantity) + total_cost += amount + details.append({"item": name, "quantity": quantity, "unit_price": price, "amount": amount}) + + return wrap_response( + { + "project_type": project_type, + "details": details, + "summary": {"total_cost_won": total_cost, "item_count": len(details)}, + }, + ProjectDomain.복합, + ) diff --git a/civilplan_mcp/updater/__init__.py b/civilplan_mcp/updater/__init__.py new file mode 100644 index 0000000..a8470c8 --- /dev/null +++ b/civilplan_mcp/updater/__init__.py @@ -0,0 +1,3 @@ +from civilplan_mcp.updater.scheduler import build_scheduler + +__all__ = ["build_scheduler"] diff --git a/civilplan_mcp/updater/common.py b/civilplan_mcp/updater/common.py new file mode 100644 index 0000000..b3ce6ab --- /dev/null +++ b/civilplan_mcp/updater/common.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +from datetime import datetime +import json +from pathlib import Path +import re +from typing import Any + +import httpx + +from civilplan_mcp.config import get_settings + + +LOG_FILE_NAME = "update_log.json" + + +def flag_manual_update_required(update_type: str, message: str, data_dir: Path | None = None) -> Path: + target_dir = data_dir or get_settings().data_dir + target_dir.mkdir(parents=True, exist_ok=True) + flag_path = target_dir / f".update_required_{update_type}" + flag_path.write_text(f"{datetime.now():%Y-%m-%d} {message}", encoding="utf-8") + return flag_path + + +def clear_update_flag(update_type: str, data_dir: Path | None = None) -> None: + target_dir = data_dir or get_settings().data_dir + flag_path = target_dir / f".update_required_{update_type}" + if flag_path.exists(): + flag_path.unlink() + + +def check_update_flags(data_dir: Path | None = None) -> list[str]: + target_dir = data_dir or get_settings().data_dir + warnings: list[str] = [] + for flag in target_dir.glob(".update_required_*"): + warnings.append(flag.read_text(encoding="utf-8")) + return warnings + + +def read_update_log(data_dir: Path | None = None) -> list[dict[str, Any]]: + target_dir = data_dir or get_settings().data_dir + log_path = target_dir / LOG_FILE_NAME + if not log_path.exists(): + return [] + return json.loads(log_path.read_text(encoding="utf-8")) + + +def write_update_log(entry: dict[str, Any], data_dir: Path | None = None) -> Path: + target_dir = data_dir or get_settings().data_dir + target_dir.mkdir(parents=True, exist_ok=True) + log_path = target_dir / LOG_FILE_NAME + log_data = read_update_log(target_dir) + log_data.append(entry) + log_path.write_text(json.dumps(log_data, ensure_ascii=False, indent=2), encoding="utf-8") + return log_path + + +def fetch_source_text(url: str) -> str: + response = httpx.get( + url, + timeout=30, + headers={"User-Agent": "CivilPlan-MCP/1.0 (+https://localhost:8765/mcp)"}, + ) + response.raise_for_status() + return response.text + + +def extract_marker(text: str, patterns: list[str]) -> str | None: + for pattern in patterns: + match = re.search(pattern, text, flags=re.IGNORECASE) + if match: + return match.group(0) + return None + + +def build_log_entry( + *, + update_type: str, + period: str | None, + status: str, + source_url: str, + detail: str, + marker: str | None = None, +) -> dict[str, Any]: + return { + "timestamp": datetime.now().isoformat(timespec="seconds"), + "update_type": update_type, + "period": period, + "status": status, + "source_url": source_url, + "marker": marker, + "detail": detail, + } diff --git a/civilplan_mcp/updater/scheduler.py b/civilplan_mcp/updater/scheduler.py new file mode 100644 index 0000000..77eaed2 --- /dev/null +++ b/civilplan_mcp/updater/scheduler.py @@ -0,0 +1,20 @@ +from __future__ import annotations + +from apscheduler.schedulers.background import BackgroundScheduler +from apscheduler.triggers.cron import CronTrigger + +from civilplan_mcp.updater.standard_updater import update_standard_prices +from civilplan_mcp.updater.wage_updater import update_wage_rates +from civilplan_mcp.updater.waste_updater import update_waste_prices + + +def build_scheduler(*, start: bool = False) -> BackgroundScheduler: + scheduler = BackgroundScheduler(timezone="Asia/Seoul") + scheduler.add_job(update_wage_rates, CronTrigger(month="1", day="2", hour="9"), id="wage_h1", kwargs={"period": "상반기"}) + scheduler.add_job(update_waste_prices, CronTrigger(month="1", day="2", hour="9"), id="waste_annual") + scheduler.add_job(update_standard_prices, CronTrigger(month="1", day="2", hour="9"), id="standard_h1", kwargs={"period": "상반기"}) + scheduler.add_job(update_standard_prices, CronTrigger(month="7", day="10", hour="9"), id="standard_h2", kwargs={"period": "하반기"}) + scheduler.add_job(update_wage_rates, CronTrigger(month="9", day="2", hour="9"), id="wage_h2", kwargs={"period": "하반기"}) + if start: + scheduler.start() + return scheduler diff --git a/civilplan_mcp/updater/standard_updater.py b/civilplan_mcp/updater/standard_updater.py new file mode 100644 index 0000000..93d28b4 --- /dev/null +++ b/civilplan_mcp/updater/standard_updater.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.config import get_settings +from civilplan_mcp.updater.common import ( + build_log_entry, + clear_update_flag, + extract_marker, + fetch_source_text, + flag_manual_update_required, + write_update_log, +) + + +STANDARD_SOURCE_URL = "https://www.molit.go.kr/portal.do" +STANDARD_MARKER_PATTERNS = [ + r"표준시장단가", + r"제비율", + r"20\d{2}년", +] + + +def update_standard_prices(period: str = "상반기") -> dict[str, Any]: + data_dir = get_settings().data_dir + try: + text = fetch_source_text(STANDARD_SOURCE_URL) + marker = extract_marker(text, STANDARD_MARKER_PATTERNS) + if not marker: + detail = f"Fetched standard-cost source, but no recognizable bulletin marker was found for {period}." + flag_manual_update_required("standard", detail, data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="standard", + period=period, + status="pending_manual_review", + source_url=STANDARD_SOURCE_URL, + detail=detail, + ), + data_dir=data_dir, + ) + return {"status": "pending_manual_review", "period": period, "source_url": STANDARD_SOURCE_URL} + + clear_update_flag("standard", data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="standard", + period=period, + status="fetched", + source_url=STANDARD_SOURCE_URL, + detail="Recognized standard-cost bulletin marker from source page.", + marker=marker, + ), + data_dir=data_dir, + ) + return { + "status": "fetched", + "period": period, + "marker": marker, + "source_url": STANDARD_SOURCE_URL, + } + except Exception as exc: + detail = f"Standard update fetch failed for {period}: {exc}" + flag_manual_update_required("standard", detail, data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="standard", + period=period, + status="pending_manual_review", + source_url=STANDARD_SOURCE_URL, + detail=detail, + ), + data_dir=data_dir, + ) + return { + "status": "pending_manual_review", + "period": period, + "message": detail, + "source_url": STANDARD_SOURCE_URL, + } diff --git a/civilplan_mcp/updater/update_log.json b/civilplan_mcp/updater/update_log.json new file mode 100644 index 0000000..fe51488 --- /dev/null +++ b/civilplan_mcp/updater/update_log.json @@ -0,0 +1 @@ +[] diff --git a/civilplan_mcp/updater/wage_updater.py b/civilplan_mcp/updater/wage_updater.py new file mode 100644 index 0000000..946315d --- /dev/null +++ b/civilplan_mcp/updater/wage_updater.py @@ -0,0 +1,84 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from civilplan_mcp.config import get_settings +from civilplan_mcp.updater.common import ( + build_log_entry, + check_update_flags, + clear_update_flag, + extract_marker, + fetch_source_text, + flag_manual_update_required, + write_update_log, +) + + +WAGE_SOURCE_URL = "https://gwangju.cak.or.kr/lay1/bbs/S1T41C42/A/14/list.do" +WAGE_MARKER_PATTERNS = [ + r"20\d{2}년\s*(상반기|하반기)\s*적용\s*건설업\s*임금실태조사", + r"20\d{2}년\s*건설업\s*임금실태조사", +] + + +def update_wage_rates(period: str = "상반기") -> dict[str, Any]: + data_dir = get_settings().data_dir + try: + text = fetch_source_text(WAGE_SOURCE_URL) + marker = extract_marker(text, WAGE_MARKER_PATTERNS) + if not marker: + detail = f"Fetched wage source, but no recognizable wage bulletin marker was found for {period}." + flag_manual_update_required("wage", detail, data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="wage", + period=period, + status="pending_manual_review", + source_url=WAGE_SOURCE_URL, + detail=detail, + ), + data_dir=data_dir, + ) + return {"status": "pending_manual_review", "period": period, "source_url": WAGE_SOURCE_URL} + + clear_update_flag("wage", data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="wage", + period=period, + status="fetched", + source_url=WAGE_SOURCE_URL, + detail="Recognized wage bulletin marker from source page.", + marker=marker, + ), + data_dir=data_dir, + ) + return { + "status": "fetched", + "period": period, + "marker": marker, + "source_url": WAGE_SOURCE_URL, + } + except Exception as exc: + detail = f"Wage update fetch failed for {period}: {exc}" + flag_manual_update_required("wage", detail, data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="wage", + period=period, + status="pending_manual_review", + source_url=WAGE_SOURCE_URL, + detail=detail, + ), + data_dir=data_dir, + ) + return { + "status": "pending_manual_review", + "period": period, + "message": detail, + "source_url": WAGE_SOURCE_URL, + } + + +__all__ = ["check_update_flags", "flag_manual_update_required", "update_wage_rates"] diff --git a/civilplan_mcp/updater/waste_updater.py b/civilplan_mcp/updater/waste_updater.py new file mode 100644 index 0000000..51fc7e4 --- /dev/null +++ b/civilplan_mcp/updater/waste_updater.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +from typing import Any + +from civilplan_mcp.config import get_settings +from civilplan_mcp.updater.common import ( + build_log_entry, + clear_update_flag, + extract_marker, + fetch_source_text, + flag_manual_update_required, + write_update_log, +) + + +WASTE_SOURCE_URL = "https://www.data.go.kr/data/15052266/fileData.do" +WASTE_MARKER_PATTERNS = [ + r"20\d{2}[-./]\d{2}[-./]\d{2}", + r"20\d{2}년", +] + + +def update_waste_prices() -> dict[str, Any]: + data_dir = get_settings().data_dir + try: + text = fetch_source_text(WASTE_SOURCE_URL) + marker = extract_marker(text, WASTE_MARKER_PATTERNS) + if not marker: + detail = "Fetched waste source, but no recognizable release marker was found." + flag_manual_update_required("waste", detail, data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="waste", + period=None, + status="pending_manual_review", + source_url=WASTE_SOURCE_URL, + detail=detail, + ), + data_dir=data_dir, + ) + return {"status": "pending_manual_review", "source_url": WASTE_SOURCE_URL} + + clear_update_flag("waste", data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="waste", + period=None, + status="fetched", + source_url=WASTE_SOURCE_URL, + detail="Recognized release marker from waste source page.", + marker=marker, + ), + data_dir=data_dir, + ) + return {"status": "fetched", "marker": marker, "source_url": WASTE_SOURCE_URL} + except Exception as exc: + detail = f"Waste update fetch failed: {exc}" + flag_manual_update_required("waste", detail, data_dir=data_dir) + write_update_log( + build_log_entry( + update_type="waste", + period=None, + status="pending_manual_review", + source_url=WASTE_SOURCE_URL, + detail=detail, + ), + data_dir=data_dir, + ) + return { + "status": "pending_manual_review", + "message": detail, + "source_url": WASTE_SOURCE_URL, + } diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..96b4618 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,33 @@ +[build-system] +requires = ["setuptools>=68", "wheel"] +build-backend = "setuptools.build_meta" + +[project] +name = "civilplan-mcp" +version = "1.0.0" +description = "CivilPlan MCP server for Korean civil and building project planning." +readme = "README.md" +requires-python = ">=3.11" +license = { text = "MIT" } +authors = [ + { name = "22B Labs", email = "sinmb79@example.com" } +] +dependencies = [ + "fastmcp>=2.0.0", + "pydantic>=2.0.0", + "openpyxl>=3.1.0", + "python-docx>=1.1.0", + "svgwrite>=1.4.3", + "ezdxf>=1.3.0", + "aiosqlite>=0.20.0", + "httpx>=0.27.0", + "apscheduler>=3.10.0", + "python-dateutil>=2.9.0", +] + +[tool.setuptools.packages.find] +include = ["civilplan_mcp*"] + +[tool.pytest.ini_options] +pythonpath = ["."] +testpaths = ["tests"] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..90dd2b5 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +fastmcp>=2.0.0 +pydantic>=2.0.0 +openpyxl>=3.1.0 +python-docx>=1.1.0 +svgwrite>=1.4.3 +ezdxf>=1.3.0 +aiosqlite>=0.20.0 +httpx>=0.27.0 +apscheduler>=3.10.0 +python-dateutil>=2.9.0 +pytest>=8.0.0 diff --git a/server.py b/server.py new file mode 100644 index 0000000..510b881 --- /dev/null +++ b/server.py @@ -0,0 +1,11 @@ +""" +CivilPlan MCP Server +LICENSE: MIT +PHILOSOPHY: Hongik Ingan - reduce inequality in access to expert planning knowledge. +""" + +from civilplan_mcp.server import main + + +if __name__ == "__main__": + main() diff --git a/tests/test_base.py b/tests/test_base.py new file mode 100644 index 0000000..8bb4e7f --- /dev/null +++ b/tests/test_base.py @@ -0,0 +1,17 @@ +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools._base import wrap_response + + +def test_project_domain_contains_expected_values() -> None: + assert ProjectDomain.건축.value == "건축" + assert ProjectDomain.토목_도로.value == "토목_도로" + assert ProjectDomain.복합.value == "복합" + + +def test_wrap_response_adds_required_disclaimers() -> None: + wrapped = wrap_response({"status": "ok"}, ProjectDomain.토목_도로) + + assert wrapped["status"] == "ok" + assert "domain_note" in wrapped + assert "validity_disclaimer" in wrapped + assert wrapped["data_as_of"] == "2026년 4월 기준" diff --git a/tests/test_data_files.py b/tests/test_data_files.py new file mode 100644 index 0000000..8c8b049 --- /dev/null +++ b/tests/test_data_files.py @@ -0,0 +1,39 @@ +from pathlib import Path + +from civilplan_mcp.config import get_settings +from civilplan_mcp.db.bootstrap import bootstrap_database, load_json_data + + +def test_required_seed_files_exist() -> None: + data_dir = get_settings().data_dir + required = [ + "unit_prices_2026.json", + "legal_procedures.json", + "region_factors.json", + "road_standards.json", + "guidelines_catalog.json", + "association_prices_catalog.json", + "waste_disposal_prices_2025.json", + "indirect_cost_rates.json", + "supervision_rates.json", + "rental_benchmark.json", + "price_update_calendar.json", + "data_validity_warnings.json", + ] + + for name in required: + assert (data_dir / name).exists(), name + + +def test_load_json_data_reads_seed_file() -> None: + data = load_json_data("region_factors.json") + + assert "경기도" in data + + +def test_bootstrap_database_creates_sqlite_file(tmp_path: Path) -> None: + db_path = tmp_path / "civilplan.db" + + bootstrap_database(db_path) + + assert db_path.exists() diff --git a/tests/test_extended_tools.py b/tests/test_extended_tools.py new file mode 100644 index 0000000..575b655 --- /dev/null +++ b/tests/test_extended_tools.py @@ -0,0 +1,36 @@ +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools.bid_type_selector import select_bid_type +from civilplan_mcp.tools.guideline_fetcher import fetch_guideline_summary +from civilplan_mcp.tools.guideline_resolver import get_applicable_guidelines +from civilplan_mcp.tools.waste_estimator import estimate_waste_disposal + + +def test_get_applicable_guidelines_returns_matches() -> None: + result = get_applicable_guidelines( + domain=ProjectDomain.토목_도로, + procedure_ids=["PER-01"], + project_type="도로", + ) + + assert result["guidelines"] + + +def test_fetch_guideline_summary_returns_catalog_summary() -> None: + result = fetch_guideline_summary(guideline_id="GL-001") + + assert result["summary"]["id"] == "GL-001" + + +def test_select_bid_type_chooses_comprehensive_for_large_cost() -> None: + result = select_bid_type(total_cost_billion=350.0, domain=ProjectDomain.건축) + + assert result["recommended_type"] == "종합심사낙찰제" + + +def test_estimate_waste_disposal_returns_cost() -> None: + result = estimate_waste_disposal( + project_type="도로", + waste_items={"폐콘크리트": 100, "폐아스팔트콘크리트": 50}, + ) + + assert result["summary"]["total_cost_won"] > 0 diff --git a/tests/test_generators.py b/tests/test_generators.py new file mode 100644 index 0000000..a2d35c7 --- /dev/null +++ b/tests/test_generators.py @@ -0,0 +1,88 @@ +from pathlib import Path + +from openpyxl import load_workbook + +from civilplan_mcp.tools.boq_generator import generate_boq_excel +from civilplan_mcp.tools.doc_generator import generate_investment_doc +from civilplan_mcp.tools.drawing_generator import generate_svg_drawing +from civilplan_mcp.tools.quantity_estimator import estimate_quantities +from civilplan_mcp.tools.project_parser import parse_project +from civilplan_mcp.tools.unit_price_query import get_unit_prices + + +def _sample_inputs(tmp_path: Path) -> tuple[dict, dict, dict]: + project_spec = parse_project( + description="소로 신설 L=890m B=6m 아스콘 2차선 상하수도 경기도 둔턱지역 2026~2028" + ) + project_spec["output_dir"] = str(tmp_path) + quantities = estimate_quantities( + road_class="소로", + length_m=890, + width_m=6.0, + terrain="구릉", + pavement_type="아스콘", + include_water_supply=True, + include_sewage=True, + include_retaining_wall=True, + include_bridge=False, + bridge_length_m=0.0, + ) + unit_prices = get_unit_prices(region="경기도", year=2026) + return project_spec, quantities, unit_prices + + +def test_generate_boq_excel_creates_workbook(tmp_path: Path) -> None: + project_spec, quantities, unit_prices = _sample_inputs(tmp_path) + + result = generate_boq_excel( + project_name="소로 개설(신설) 공사", + project_spec=project_spec, + quantities=quantities, + unit_prices=unit_prices, + region="경기도", + year=2026, + output_filename="boq.xlsx", + ) + + workbook = load_workbook(result["file_path"], data_only=False) + assert "사업내역서(BOQ)" in workbook.sheetnames + + +def test_generate_investment_doc_creates_docx(tmp_path: Path) -> None: + project_spec, quantities, unit_prices = _sample_inputs(tmp_path) + boq = generate_boq_excel( + project_name="소로 개설(신설) 공사", + project_spec=project_spec, + quantities=quantities, + unit_prices=unit_prices, + region="경기도", + year=2026, + output_filename="boq_for_doc.xlsx", + ) + + result = generate_investment_doc( + project_name="소로 개설(신설) 공사", + project_spec=project_spec, + quantities=quantities, + legal_procedures={"summary": {"total_procedures": 3}, "phases": {}}, + boq_summary=boq["summary"], + requester="22B Labs", + output_filename="investment.docx", + ) + + assert Path(result["file_path"]).exists() + + +def test_generate_svg_drawing_creates_svg(tmp_path: Path) -> None: + project_spec, quantities, _ = _sample_inputs(tmp_path) + + result = generate_svg_drawing( + drawing_type="횡단면도", + project_spec=project_spec, + quantities=quantities, + scale="1:200", + output_filename="road.svg", + ) + + content = Path(result["file_path"]).read_text(encoding="utf-8") + assert " None: + result = evaluate_impact_assessments( + domain="토목_도로", + project_type="도로", + road_length_m=890, + development_area_m2=5340, + total_cost_billion=10.67, + building_floor_area_m2=None, + housing_units=None, + is_urban_area=True, + near_cultural_heritage=False, + near_river=False, + near_protected_area=False, + ) + + target = next(item for item in result["evaluations"] if item["name"] == "재해영향평가") + assert target["result"] == "BORDERLINE" diff --git a/tests/test_land_info_and_benchmark_fallbacks.py b/tests/test_land_info_and_benchmark_fallbacks.py new file mode 100644 index 0000000..9891658 --- /dev/null +++ b/tests/test_land_info_and_benchmark_fallbacks.py @@ -0,0 +1,113 @@ +from __future__ import annotations + +import httpx + +from civilplan_mcp.config import Settings +from civilplan_mcp.tools import benchmark_validator, land_info_query + + +def _json_response(url: str, payload: dict, *, status_code: int = 200) -> httpx.Response: + return httpx.Response( + status_code, + request=httpx.Request("GET", url), + json=payload, + ) + + +def test_query_land_info_uses_eum_land_use_endpoint(monkeypatch, tmp_path) -> None: + settings = Settings(vworld_api_key="test-key", data_go_kr_api_key="public-key", data_dir=tmp_path) + calls: list[str] = [] + + def fake_get(url: str, *, params=None, timeout=None, headers=None): # type: ignore[no-untyped-def] + calls.append(url) + if url == land_info_query.VWORLD_ADDRESS_URL: + return _json_response( + url, + { + "response": { + "status": "OK", + "result": { + "point": {"x": "127.061", "y": "37.821"}, + "items": [ + { + "id": "4163011600101230000", + "address": {"parcel": "경기도 양주시 덕계동 123"}, + } + ], + }, + } + }, + ) + if url == land_info_query.VWORLD_DATA_URL: + return _json_response( + url, + { + "response": { + "status": "OK", + "result": { + "featureCollection": { + "features": [ + { + "properties": { + "area": "320.5", + "jimok": "대", + } + } + ] + } + }, + } + }, + ) + if url == land_info_query.EUM_LAND_USE_URL: + return _json_response( + url, + { + "landUseInfo": { + "useDistrict": "제2종일반주거지역", + "district2": "고도지구", + "bcr": "60", + "far": "200", + "heightLimit": "16", + } + }, + ) + raise AssertionError(f"Unexpected URL: {url}") + + monkeypatch.setattr(land_info_query, "get_settings", lambda: settings) + monkeypatch.setattr(land_info_query.httpx, "get", fake_get) + + result = land_info_query.query_land_info(address="경기도 양주시 덕계동 123", pnu=None) + + assert result["status"] == "success" + assert result["zoning"]["use_district"] == "제2종일반주거지역" + assert result["zoning"]["district_2"] == "고도지구" + assert result["zoning"]["bcr_pct"] == 60 + assert result["zoning"]["far_pct"] == 200 + assert land_info_query.EUM_LAND_USE_URL in calls + assert "https://api.vworld.kr/req/wfs" not in calls + + +def test_validate_against_benchmark_reports_backend_fallback_on_502(monkeypatch) -> None: + settings = Settings(data_go_kr_api_key="test-key") + + def fake_get(url: str, *, params=None, timeout=None): # type: ignore[no-untyped-def] + return httpx.Response( + 502, + request=httpx.Request("GET", url), + text="backend unavailable", + ) + + monkeypatch.setattr(benchmark_validator, "get_settings", lambda: settings) + monkeypatch.setattr(benchmark_validator.httpx, "get", fake_get) + + result = benchmark_validator.validate_against_benchmark( + project_type="도로_포장", + road_length_m=890, + floor_area_m2=None, + region="경기도", + our_estimate_won=1_067_000_000, + ) + + assert result["benchmark"]["api_status"] == "fallback" + assert "나라장터 직접 검색" in result["benchmark"]["availability_note"] diff --git a/tests/test_land_info_ingestion_and_updater.py b/tests/test_land_info_ingestion_and_updater.py new file mode 100644 index 0000000..ef48e9b --- /dev/null +++ b/tests/test_land_info_ingestion_and_updater.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from io import BytesIO +from pathlib import Path +import zipfile + +from civilplan_mcp.config import Settings +from civilplan_mcp.tools.land_info_query import ( + _read_land_price_from_files, + extract_land_use_html_properties, +) +from civilplan_mcp.updater import wage_updater + + +def test_extract_land_use_html_properties_reads_regulation_table() -> None: + html = """ +
+ + + + + + + + + + + + +
제2종일반주거지역60%200%
건폐율

최대 건폐율 60%

용적률

최대 용적률 200%

+
+ """ + + parsed = extract_land_use_html_properties(html) + + assert parsed["usedistrictnm"] == "제2종일반주거지역" + assert parsed["bcr"] == "60" + assert parsed["far"] == "200" + + +def test_read_land_price_from_zip_file(tmp_path: Path, monkeypatch) -> None: + land_price_dir = tmp_path / "land_prices" + land_price_dir.mkdir() + archive_path = land_price_dir / "seoul_land_prices.zip" + + buffer = BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + archive.writestr( + "prices.csv", + "PNU,LANDPRICE\n1111010100100010000,1250000\n", + ) + archive_path.write_bytes(buffer.getvalue()) + + settings = Settings(data_dir=tmp_path) + monkeypatch.setattr("civilplan_mcp.tools.land_info_query.get_settings", lambda: settings) + + price = _read_land_price_from_files("1111010100100010000") + + assert price == {"individual_m2_won": 1250000, "source": "seoul_land_prices.zip:prices.csv"} + + +def test_update_wage_rates_creates_flag_and_log_on_parse_failure(tmp_path: Path, monkeypatch) -> None: + settings = Settings(data_dir=tmp_path) + + monkeypatch.setattr(wage_updater, "get_settings", lambda: settings) + monkeypatch.setattr( + wage_updater, + "fetch_source_text", + lambda url: "No structured wage table here", + ) + + result = wage_updater.update_wage_rates(period="상반기") + + assert result["status"] == "pending_manual_review" + assert (tmp_path / ".update_required_wage").exists() + assert (tmp_path / "update_log.json").exists() diff --git a/tests/test_legal.py b/tests/test_legal.py new file mode 100644 index 0000000..24f86b5 --- /dev/null +++ b/tests/test_legal.py @@ -0,0 +1,34 @@ +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools.legal_procedures import get_legal_procedures +from civilplan_mcp.tools.phase_checklist import get_phase_checklist + + +def test_get_legal_procedures_returns_domain_specific_items() -> None: + result = get_legal_procedures( + domain=ProjectDomain.토목_도로, + project_type="도로", + total_cost_billion=10.67, + road_length_m=890, + development_area_m2=5340, + region="경기도", + has_farmland=False, + has_forest=False, + has_river=False, + is_public=True, + ) + + assert result["summary"]["total_procedures"] >= 2 + assert "인허가" in result["phases"] + + +def test_get_phase_checklist_returns_preparing_message_for_landscape() -> None: + result = get_phase_checklist( + domain=ProjectDomain.조경, + phase="기획", + project_type="조경", + total_cost_billion=3.0, + has_building=False, + has_bridge=False, + ) + + assert result["status"] == "not_ready" diff --git a/tests/test_parser.py b/tests/test_parser.py new file mode 100644 index 0000000..a3f20bf --- /dev/null +++ b/tests/test_parser.py @@ -0,0 +1,21 @@ +from civilplan_mcp.models import ProjectDomain +from civilplan_mcp.tools.project_parser import parse_project + + +def test_parse_project_extracts_core_road_fields() -> None: + result = parse_project( + description="소로 신설 L=890m B=6m 아스콘 2차선 상하수도 경기도 둔턱지역 2026~2028" + ) + + assert result["road"]["length_m"] == 890 + assert result["road"]["width_m"] == 6.0 + assert result["road"]["lanes"] == 2 + assert result["region"] == "경기도" + assert result["domain"] == ProjectDomain.토목_도로.value + + +def test_parse_project_marks_composite_domain_warning() -> None: + result = parse_project(description="복지관 신축 및 진입도로 개설") + + assert result["domain"] == ProjectDomain.복합.value + assert "domain_warning" in result diff --git a/tests/test_quantities.py b/tests/test_quantities.py new file mode 100644 index 0000000..e3226fc --- /dev/null +++ b/tests/test_quantities.py @@ -0,0 +1,27 @@ +from civilplan_mcp.tools.quantity_estimator import estimate_quantities +from civilplan_mcp.tools.unit_price_query import get_unit_prices + + +def test_estimate_quantities_returns_expected_pavement_area() -> None: + result = estimate_quantities( + road_class="소로", + length_m=890, + width_m=6.0, + terrain="구릉", + pavement_type="아스콘", + include_water_supply=True, + include_sewage=True, + include_retaining_wall=True, + include_bridge=False, + bridge_length_m=0.0, + ) + + assert result["road_spec"]["pavement_area_m2"] == 5340.0 + assert result["quantities"]["포장"]["아스콘표층_t"] > 0 + + +def test_get_unit_prices_applies_region_factor() -> None: + result = get_unit_prices(category="포장", item_name="아스콘", region="경기도", year=2026) + + assert result["results"] + assert result["results"][0]["adjusted_price"] > result["results"][0]["base_price"] diff --git a/tests/test_server_registration.py b/tests/test_server_registration.py new file mode 100644 index 0000000..3434cf2 --- /dev/null +++ b/tests/test_server_registration.py @@ -0,0 +1,29 @@ +import asyncio + +from civilplan_mcp.server import build_mcp, build_server_config + + +def test_build_server_config_defaults() -> None: + config = build_server_config() + + assert config["host"] == "127.0.0.1" + assert config["port"] == 8765 + assert config["path"] == "/mcp" + + +def test_server_registers_all_19_tools() -> None: + app = build_mcp() + tools = asyncio.run(app.list_tools()) + names = {tool.name for tool in tools} + + assert len(names) == 19 + assert "civilplan_parse_project" in names + assert "civilplan_generate_dxf_drawing" in names + + +def test_read_tools_have_read_only_hint() -> None: + app = build_mcp() + tools = {tool.name: tool for tool in asyncio.run(app.list_tools())} + + assert tools["civilplan_parse_project"].annotations.readOnlyHint is True + assert tools["civilplan_generate_boq_excel"].annotations.readOnlyHint is None diff --git a/tests/test_smoke.py b/tests/test_smoke.py new file mode 100644 index 0000000..8ceb083 --- /dev/null +++ b/tests/test_smoke.py @@ -0,0 +1,18 @@ +from civilplan_mcp import __version__ +from civilplan_mcp.config import Settings, get_settings + + +def test_package_version_present() -> None: + assert __version__ == "1.0.0" + + +def test_settings_have_expected_defaults() -> None: + settings = Settings() + + assert settings.host == "127.0.0.1" + assert settings.port == 8765 + assert settings.http_path == "/mcp" + + +def test_get_settings_is_cached() -> None: + assert get_settings() is get_settings() diff --git a/tests/test_updater.py b/tests/test_updater.py new file mode 100644 index 0000000..68bb4f1 --- /dev/null +++ b/tests/test_updater.py @@ -0,0 +1,23 @@ +from pathlib import Path + +from civilplan_mcp.updater.scheduler import build_scheduler +from civilplan_mcp.updater.wage_updater import check_update_flags, flag_manual_update_required + + +def test_build_scheduler_registers_expected_jobs() -> None: + scheduler = build_scheduler(start=False) + + assert {job.id for job in scheduler.get_jobs()} == { + "wage_h1", + "waste_annual", + "standard_h1", + "standard_h2", + "wage_h2", + } + + +def test_flag_manual_update_required_creates_flag(tmp_path: Path) -> None: + flag_manual_update_required("wage", "manual review needed", data_dir=tmp_path) + + warnings = check_update_flags(data_dir=tmp_path) + assert warnings diff --git a/tests/test_v11_tools.py b/tests/test_v11_tools.py new file mode 100644 index 0000000..e26f0d5 --- /dev/null +++ b/tests/test_v11_tools.py @@ -0,0 +1,159 @@ +from pathlib import Path + +import ezdxf + +from civilplan_mcp.config import Settings +from civilplan_mcp.tools.benchmark_validator import validate_against_benchmark +from civilplan_mcp.tools.budget_report_generator import generate_budget_report +from civilplan_mcp.tools.dxf_generator import generate_dxf_drawing +from civilplan_mcp.tools.feasibility_analyzer import analyze_feasibility +from civilplan_mcp.tools.land_info_query import ( + build_land_use_bbox_params, + extract_address_result, + extract_feature_properties, + query_land_info, +) +from civilplan_mcp.tools.project_parser import parse_project +from civilplan_mcp.tools.quantity_estimator import estimate_quantities + + +def test_query_land_info_returns_graceful_message_without_api_keys() -> None: + result = query_land_info(address="경기도 양주시 덕계동 123", pnu=None) + + assert result["status"] == "disabled" + + +def test_settings_keep_public_data_key_for_nara_usage() -> None: + settings = Settings(data_go_kr_api_key="abc") + + assert settings.data_go_kr_api_key == "abc" + + +def test_extract_address_result_reads_vworld_shape() -> None: + payload = { + "response": { + "status": "OK", + "result": { + "point": {"x": "127.061", "y": "37.821"}, + "items": [ + { + "id": "4163010100101230000", + "address": {"parcel": "경기도 양주시 덕계동 123"} + } + ], + }, + } + } + + parsed = extract_address_result(payload) + assert parsed["pnu"] == "4163010100101230000" + assert parsed["x"] == 127.061 + + +def test_extract_feature_properties_returns_first_feature() -> None: + payload = { + "response": { + "status": "OK", + "result": { + "featureCollection": { + "features": [ + {"properties": {"pnu": "4163010100101230000", "jimok": "대"}} + ] + } + }, + } + } + + parsed = extract_feature_properties(payload) + assert parsed["jimok"] == "대" + + +def test_build_land_use_bbox_params_uses_wfs_bbox_pattern() -> None: + params = build_land_use_bbox_params(127.0, 37.0, "test-key") + + assert params["SERVICE"] == "WFS" + assert params["REQUEST"] == "GetFeature" + assert params["TYPENAME"] == "lt_c_lhblpn" + assert params["SRSNAME"] == "EPSG:4326" + assert params["KEY"] == "test-key" + assert params["OUTPUTFORMAT"] == "application/json" + assert params["BBOX"] == "126.9995,36.9995,127.0005,37.0005,EPSG:4326" + + +def test_analyze_feasibility_returns_positive_cost_structure() -> None: + result = analyze_feasibility( + land_area_m2=600, + land_price_per_m2=850000, + land_price_multiplier=1.0, + construction_cost_total=890000000, + other_costs_million=50, + revenue_type="임대", + building_floor_area_m2=720, + sale_price_per_m2=None, + monthly_rent_per_m2=16000, + vacancy_rate_pct=10.0, + operating_expense_pct=20.0, + equity_ratio_pct=30.0, + loan_rate_pct=5.5, + loan_term_years=10, + construction_months=24, + sale_months=12, + ) + + assert result["cost_structure"]["total_investment"] > 0 + assert "irr_pct" in result["returns"] + + +def test_validate_against_benchmark_includes_bid_warning() -> None: + result = validate_against_benchmark( + project_type="소로_도로", + road_length_m=890, + floor_area_m2=None, + region="경기도", + our_estimate_won=1067000000, + ) + + assert "낙찰가 ≠ 사업비" in result["bid_rate_reference"]["경고"] + + +def test_generate_budget_report_creates_docx(tmp_path: Path) -> None: + project_data = parse_project(description="복지관 신축 2026~2028 경기도") + project_data["output_dir"] = str(tmp_path) + result = generate_budget_report( + report_type="예산편성요구서", + project_data=project_data, + boq_summary={"total_cost": 1067000000, "direct_cost": 900422000, "indirect_cost": 166578000}, + department="도로과", + requester="22B Labs", + output_filename="budget.docx", + ) + + assert Path(result["file_path"]).exists() + + +def test_generate_dxf_drawing_creates_dxf(tmp_path: Path) -> None: + project_spec = parse_project(description="소로 신설 L=890m B=6m 경기도") + project_spec["output_dir"] = str(tmp_path) + quantities = estimate_quantities( + road_class="소로", + length_m=890, + width_m=6.0, + terrain="평지", + pavement_type="아스콘", + include_water_supply=False, + include_sewage=False, + include_retaining_wall=False, + include_bridge=False, + bridge_length_m=0.0, + ) + + result = generate_dxf_drawing( + drawing_type="횡단면도", + project_spec=project_spec, + quantities=quantities, + scale="1:200", + output_filename="road.dxf", + ) + + doc = ezdxf.readfile(result["file_path"]) + assert doc is not None