Update linked code labels and alias cache

mysql_preview_server.py

@@ -12,6 +12,7 @@ import html as html_lib
 import threading
 import uuid
 import subprocess
+from pathlib import Path
 from datetime import datetime, timedelta, time
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from urllib.parse import parse_qs, quote, urlparse
@@ -140,6 +141,14 @@ def init_db(conn):
             shortName TEXT
         );
 
+        CREATE TABLE IF NOT EXISTS erp_project_alias_cache (
+            sourcePage TEXT NOT NULL,
+            projectCode TEXT NOT NULL,
+            shortName TEXT NOT NULL DEFAULT '',
+            syncedAt TEXT NOT NULL,
+            PRIMARY KEY (sourcePage, projectCode)
+        );
+
         CREATE TABLE IF NOT EXISTS site_worksheet_record (
             projectCode TEXT,
             workDate TEXT,
@@ -288,6 +297,20 @@ def init_db(conn):
             PRIMARY KEY (sourcePage, projectCode)
         );
 
+        CREATE TABLE IF NOT EXISTS erp_linked_code_cache (
+            sourcePage TEXT NOT NULL,
+            projectCode TEXT NOT NULL,
+            projectName TEXT NOT NULL DEFAULT '',
+            businessCode TEXT DEFAULT '',
+            salesCode TEXT DEFAULT '',
+            salesName TEXT DEFAULT '',
+            designCode TEXT DEFAULT '',
+            designName TEXT DEFAULT '',
+            matchedBy TEXT DEFAULT '',
+            syncedAt TEXT NOT NULL,
+            PRIMARY KEY (sourcePage, projectCode)
+        );
+
         CREATE TABLE IF NOT EXISTS naver_address_token_cache (
             address TEXT PRIMARY KEY,
             tokenX TEXT NOT NULL,
@@ -328,6 +351,7 @@ def init_db(conn):
         conn.execute("ALTER TABLE dallyproject ADD COLUMN BusinessTripHours REAL DEFAULT 0")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_dally_workdate ON dallyproject(WorkDate)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_project_alias_code ON project_alias(projectCode)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_erp_project_alias_cache_page ON erp_project_alias_cache(sourcePage, projectCode)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_site_worksheet_member ON site_worksheet_record(memberNo, workDate)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_site_worksheet_name ON site_worksheet_record(korName, workDate)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_site_worksheet_sync_member ON site_worksheet_sync(memberNo, yearMonth)")
@@ -340,6 +364,7 @@ def init_db(conn):
     conn.execute("CREATE INDEX IF NOT EXISTS idx_erp_contract_detail_cache_page ON erp_contract_detail_cache(sourcePage, projectCode)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_erp_bridge_overview_cache_page ON erp_bridge_overview_cache(sourcePage, projectCode, bridgeNo)")
     conn.execute("CREATE INDEX IF NOT EXISTS idx_erp_budget_plan_cache_page ON erp_budget_plan_cache(sourcePage, projectCode)")
+    conn.execute("CREATE INDEX IF NOT EXISTS idx_erp_linked_code_cache_page ON erp_linked_code_cache(sourcePage, projectCode)")
     conn.commit()
     DB_SCHEMA_READY = True
 
@@ -853,7 +878,7 @@ def backfill_daily_hours_if_needed(conn):
     return need
 
 
-def load_project_alias_from_erp():
+def load_project_alias_entries_from_erp():
     login_url = ERP_BASE_URL.rstrip('/') + '/sys/controller/common/main_controller.php'
     s = requests.Session()
     login_payload = {
@@ -886,8 +911,12 @@ def load_project_alias_from_erp():
             code = _as_text(item.get('ProjectCode')).strip().upper()
             nick = _as_text(item.get('ProjectNickname')).strip()
             if code and nick:
-                out[code] = nick
-    return sorted(out.items())
+                out[code] = {'sourcePage': page, 'projectCode': code, 'shortName': nick}
+    return sorted(out.values(), key=lambda item: (item['projectCode'], item['sourcePage']))
+
+
+def load_project_alias_from_erp():
+    return [(item['projectCode'], item['shortName']) for item in load_project_alias_entries_from_erp()]
 
 
 def erp_login_session():
@@ -1679,6 +1708,7 @@ def fetch_erp_contract_detail(page='const', project_code='', project_name=''):
         'finalContractAmountText': contract_map.get('최종계약금액', ''),
         'finalContractAmountValue': _parse_amount_text(contract_map.get('최종계약금액', '')),
         'contractType': contract_map.get('계약종류', ''),
+        'linkedCodes': find_erp_linked_codes(code, contract_map.get('사업코드', ''), short_name or contract_map.get('약칭', '')),
         'contractFields': contract_map,
         'scaleRows': scale_rows,
     }
@@ -1695,6 +1725,7 @@ def replace_erp_contract_detail_cache(conn, detail):
     raw_payload = {
         'contractFields': contract_fields,
         'scaleRows': detail.get('scaleRows') or [],
+        'linkedCodes': detail.get('linkedCodes') or {},
     }
 
     conn.execute(
@@ -1743,9 +1774,11 @@ def get_erp_contract_detail_cache(conn, source_page='const', project_code=''):
     if isinstance(raw_contract, dict) and ('contractFields' in raw_contract or 'scaleRows' in raw_contract):
         contract_fields = raw_contract.get('contractFields') or {}
         scale_rows = raw_contract.get('scaleRows') or []
+        linked_codes = _decorate_linked_code_names(raw_contract.get('linkedCodes') or {}, conn)
     else:
         contract_fields = raw_contract if isinstance(raw_contract, dict) else {}
         scale_rows = []
+        linked_codes = {}
     return {
         'sourcePage': source_page,
         'projectCode': project_code,
@@ -1757,6 +1790,7 @@ def get_erp_contract_detail_cache(conn, source_page='const', project_code=''):
         'finalContractAmountText': row[5] or '',
         'finalContractAmountValue': row[6] or 0,
         'contractType': row[7] or '',
+        'linkedCodes': linked_codes,
         'contractFields': contract_fields,
         'scaleRows': scale_rows,
         'syncedAt': row[9] or '',
@@ -1768,6 +1802,156 @@ def _clean_html_text(value):
     return re.sub(r'\s+', ' ', text).strip()
 
 
+def _normalize_text_key(value):
+    return re.sub(r'\s+', '', _as_text(value)).strip()
+
+
+def _extract_business_code_key(value):
+    text = _as_text(value).strip()
+    match = re.search(r'\d{2}-\d{3}', text)
+    return match.group(0) if match else text
+
+
+def _load_linked_code_map_from_html():
+    html_path = Path(__file__).with_name('project-codes.html')
+    if not html_path.exists():
+        return {}
+    try:
+        html_text = html_path.read_text(encoding='utf-8')
+    except OSError:
+        return {}
+    match = re.search(r'const\s+LINKED_CODE_BY_BUSINESS\s*=\s*(\{.*?\});', html_text, re.S)
+    if not match:
+        return {}
+    try:
+        parsed = json.loads(match.group(1))
+    except Exception:
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
+def _decorate_linked_code_names(linked_codes, conn=None):
+    payload = dict(linked_codes or {})
+    close_conn = False
+    if conn is None:
+        conn = open_db_connection()
+        close_conn = True
+    try:
+        sales_code = _as_text(payload.get('salesCode')).strip()
+        design_code = _as_text(payload.get('designCode')).strip()
+        if sales_code and not _as_text(payload.get('salesName')).strip():
+            payload['salesName'] = get_project_alias_name(conn, sales_code, 'sales')
+        if design_code and not _as_text(payload.get('designName')).strip():
+            payload['designName'] = get_project_alias_name(conn, design_code, 'design')
+        return payload
+    finally:
+        if close_conn:
+            conn.close()
+
+
+def fetch_erp_linked_code_rows():
+    session = erp_login_session()
+    report_url = ERP_BASE_URL.rstrip('/') + '/sys/controller/report/design_step_controller.php'
+    response = session.get(report_url, params={'ActionMode': 'REPORT_5'}, timeout=20)
+    html = response.content.decode('utf-8', errors='ignore')
+    tables = _extract_table_rows(html)
+    target_rows = []
+    for rows in tables:
+        if not rows:
+            continue
+        header = rows[0]
+        if '사업코드' in header and '영업코드' in header and '설계코드' in header and '시공코드' in header:
+            target_rows = rows
+            break
+    if not target_rows:
+        return []
+
+    header = target_rows[0]
+    header_index = {label: index for index, label in enumerate(header)}
+    business_index = header_index.get('사업코드')
+    bridge_name_index = header_index.get('교량명')
+    sales_index = header_index.get('영업코드')
+    design_index = header_index.get('설계코드')
+    const_index = header_index.get('시공코드')
+    status_index = header_index.get('진행상태')
+
+    out = []
+    for row in target_rows[1:]:
+        if not row:
+            continue
+        if business_index is None or business_index >= len(row):
+            continue
+        business_code = _as_text(row[business_index]).strip()
+        if not business_code or not re.match(r'^\d{2}-\d{3}', business_code):
+            continue
+        out.append({
+            'businessCode': business_code,
+            'bridgeName': _as_text(row[bridge_name_index]).strip() if bridge_name_index is not None and bridge_name_index < len(row) else '',
+            'status': _as_text(row[status_index]).strip() if status_index is not None and status_index < len(row) else '',
+            'salesCode': _as_text(row[sales_index]).strip() if sales_index is not None and sales_index < len(row) else '',
+            'designCode': _as_text(row[design_index]).strip() if design_index is not None and design_index < len(row) else '',
+            'constCode': _as_text(row[const_index]).strip() if const_index is not None and const_index < len(row) else '',
+        })
+    return out
+
+
+def find_erp_linked_codes(project_code='', business_code='', project_name=''):
+    project_code = _as_text(project_code).strip()
+    business_code = _extract_business_code_key(business_code)
+    project_name_key = _normalize_text_key(project_name)
+    html_map = _load_linked_code_map_from_html()
+
+    def html_map_match():
+        if project_code:
+            for key, value in html_map.items():
+                if _as_text((value or {}).get('constCode')).strip() == project_code:
+                    return _decorate_linked_code_names({
+                        'businessCode': key,
+                        'bridgeName': '',
+                        'status': '',
+                        'salesCode': _as_text((value or {}).get('salesCode')).strip(),
+                        'designCode': _as_text((value or {}).get('designCode')).strip(),
+                        'constCode': _as_text((value or {}).get('constCode')).strip(),
+                        'matchedBy': 'htmlMap.constCode',
+                    })
+        if business_code and business_code in html_map:
+            value = html_map.get(business_code) or {}
+            return _decorate_linked_code_names({
+                'businessCode': business_code,
+                'bridgeName': '',
+                'status': '',
+                'salesCode': _as_text(value.get('salesCode')).strip(),
+                'designCode': _as_text(value.get('designCode')).strip(),
+                'constCode': _as_text(value.get('constCode')).strip(),
+                'matchedBy': 'htmlMap.businessCode',
+            })
+        return None
+
+    html_hit = html_map_match()
+    if html_hit:
+        return html_hit
+
+    rows = fetch_erp_linked_code_rows()
+    if not rows:
+        return {'salesCode': '', 'salesName': '', 'designCode': '', 'designName': '', 'constCode': '', 'matchedBy': ''}
+
+    for row in rows:
+        if project_code and _as_text(row.get('constCode')).strip() == project_code:
+            return _decorate_linked_code_names({**row, 'matchedBy': 'constCode'})
+
+    candidates = [row for row in rows if business_code and _as_text(row.get('businessCode')).strip() == business_code]
+    if len(candidates) == 1:
+        return _decorate_linked_code_names({**candidates[0], 'matchedBy': 'businessCode'})
+    if len(candidates) > 1 and project_name_key:
+        for row in candidates:
+            bridge_name_key = _normalize_text_key(row.get('bridgeName'))
+            if project_name_key and project_name_key in bridge_name_key:
+                return _decorate_linked_code_names({**row, 'matchedBy': 'businessCode+projectName'})
+        return _decorate_linked_code_names({**candidates[0], 'matchedBy': 'businessCode'})
+
+    return {'salesCode': '', 'salesName': '', 'designCode': '', 'designName': '', 'constCode': '', 'matchedBy': ''}
+
+
 def _parse_site_worker_rows(html):
     rows = []
     for table in re.findall(r'<table[^>]*>(.*?)</table>', html or '', re.I | re.S):
@@ -2726,11 +2910,152 @@ def replace_project_alias(conn, rows):
     conn.commit()
 
 
+def replace_erp_project_alias_cache(conn, rows):
+    synced_at = datetime.now().isoformat(timespec='seconds')
+    conn.execute('DELETE FROM erp_project_alias_cache')
+    if rows:
+        conn.executemany(
+            '''
+            INSERT OR REPLACE INTO erp_project_alias_cache (sourcePage, projectCode, shortName, syncedAt)
+            VALUES (?, ?, ?, ?)
+            ''',
+            [
+                (
+                    _as_text(row.get('sourcePage')).strip().lower(),
+                    _as_text(row.get('projectCode')).strip().upper(),
+                    _as_text(row.get('shortName')).strip(),
+                    synced_at,
+                )
+                for row in rows
+                if _as_text(row.get('sourcePage')).strip() and _as_text(row.get('projectCode')).strip()
+            ],
+        )
+    conn.commit()
+    return {'count': len(rows), 'syncedAt': synced_at}
+
+
 def get_project_alias_map(conn):
     cur = conn.execute('SELECT projectCode, shortName FROM project_alias')
     return {code: short for code, short in cur.fetchall() if code}
 
 
+def infer_source_page_from_code(project_code=''):
+    code = _as_text(project_code).strip().upper()
+    if '-교영-' in code:
+        return 'sales'
+    if '-설계-' in code:
+        return 'design'
+    if '-시공-' in code:
+        return 'const'
+    if '-제조-' in code:
+        return 'make'
+    if '-연구-' in code:
+        return 'research'
+    return ''
+
+
+def get_project_alias_name(conn, project_code='', source_page=''):
+    code = _as_text(project_code).strip().upper()
+    page_name = _as_text(source_page).strip().lower() or infer_source_page_from_code(code)
+    if not code:
+        return ''
+    if page_name:
+        row = conn.execute(
+            'SELECT shortName FROM erp_project_alias_cache WHERE sourcePage = ? AND projectCode = ?',
+            (page_name, code),
+        ).fetchone()
+        if row and row[0]:
+            return row[0]
+    row = conn.execute('SELECT shortName FROM project_alias WHERE projectCode = ?', (code,)).fetchone()
+    return row[0] if row and row[0] else ''
+
+
+def replace_erp_linked_code_cache(conn, source_page, project_code, project_name, business_code, linked_codes):
+    source_page = _as_text(source_page).strip().lower() or 'const'
+    project_code = _as_text(project_code).strip()
+    if not project_code:
+        return None
+    payload = linked_codes or {}
+    synced_at = datetime.now().isoformat(timespec='seconds')
+    conn.execute(
+        '''
+        INSERT OR REPLACE INTO erp_linked_code_cache
+        (sourcePage, projectCode, projectName, businessCode, salesCode, salesName, designCode, designName, matchedBy, syncedAt)
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+        ''',
+        (
+            source_page,
+            project_code,
+            _as_text(project_name).strip(),
+            _extract_business_code_key(business_code),
+            _as_text(payload.get('salesCode')).strip(),
+            _as_text(payload.get('salesName')).strip(),
+            _as_text(payload.get('designCode')).strip(),
+            _as_text(payload.get('designName')).strip(),
+            _as_text(payload.get('matchedBy')).strip(),
+            synced_at,
+        ),
+    )
+    conn.commit()
+    return {'sourcePage': source_page, 'projectCode': project_code, 'syncedAt': synced_at}
+
+
+def sync_project_alias_caches(conn):
+    alias_entries = load_project_alias_entries_from_erp()
+    replace_project_alias(conn, [(item['projectCode'], item['shortName']) for item in alias_entries])
+    cache_info = replace_erp_project_alias_cache(conn, alias_entries)
+    return {'entries': alias_entries, 'cacheInfo': cache_info}
+
+
+def rebuild_erp_linked_code_cache(conn, source_page='const'):
+    source_page = _as_text(source_page).strip().lower() or 'const'
+    rows = conn.execute(
+        '''
+        SELECT p.projectCode,
+               p.projectName,
+               COALESCE(d.businessCode, '')
+        FROM erp_project_code_cache p
+        LEFT JOIN erp_contract_detail_cache d
+          ON d.sourcePage = p.sourcePage
+         AND d.projectCode = p.projectCode
+        WHERE p.sourcePage = ?
+        ORDER BY p.projectCode ASC
+        ''',
+        (source_page,),
+    ).fetchall()
+    synced_at = datetime.now().isoformat(timespec='seconds')
+    payload_rows = []
+    for project_code, project_name, business_code in rows:
+        linked_codes = find_erp_linked_codes(project_code, business_code, project_name)
+        if linked_codes.get('salesCode') or linked_codes.get('designCode'):
+            payload_rows.append(
+                (
+                    source_page,
+                    _as_text(project_code).strip(),
+                    _as_text(project_name).strip(),
+                    _extract_business_code_key(business_code),
+                    _as_text(linked_codes.get('salesCode')).strip(),
+                    _as_text(linked_codes.get('salesName')).strip(),
+                    _as_text(linked_codes.get('designCode')).strip(),
+                    _as_text(linked_codes.get('designName')).strip(),
+                    _as_text(linked_codes.get('matchedBy')).strip(),
+                    synced_at,
+                )
+            )
+    conn.execute('DELETE FROM erp_linked_code_cache WHERE sourcePage = ?', (source_page,))
+    if payload_rows:
+        conn.executemany(
+            '''
+            INSERT OR REPLACE INTO erp_linked_code_cache
+            (sourcePage, projectCode, projectName, businessCode, salesCode, salesName, designCode, designName, matchedBy, syncedAt)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            ''',
+            payload_rows,
+        )
+    conn.commit()
+    return len(payload_rows)
+
+
 def rebuild_work_calendar_tables(conn):
     conn.execute('DELETE FROM work_calendar_detail')
     conn.execute('DELETE FROM work_calendar_day')
@@ -3382,7 +3707,8 @@ def _duration_sql(alias='d'):
 
 
 def get_people_summary(conn, start_date, end_date):
-    q = f'''
+    params = (start_date, end_date, start_date, end_date)
+    q_with_site = f'''
     WITH d AS (
       SELECT
         MemberNo,
@@ -3417,7 +3743,37 @@ def get_people_summary(conn, start_date, end_date):
     LEFT JOIN s ON s.MemberNo = m.MemberNo
     ORDER BY CASE WHEN IFNULL(m.korName,'')='' THEN 1 ELSE 0 END, korName ASC, m.MemberNo ASC
     '''
-    cur = conn.execute(q, (start_date, end_date, start_date, end_date))
+    try:
+        cur = conn.execute(q_with_site, params)
+    except sqlite3.DatabaseError as error:
+        if 'malformed' not in str(error).lower():
+            raise
+        q_without_site = f'''
+        WITH d AS (
+          SELECT
+            MemberNo,
+            ROUND(SUM(TotalHours), 2) AS totalHours,
+            COUNT(*) AS totalRows
+          FROM dallyproject
+          WHERE date(substr(EntryTime,1,10)) BETWEEN date(?) AND date(?)
+          GROUP BY MemberNo
+        )
+        SELECT
+          m.MemberNo,
+          IFNULL(m.korName, '') AS korName,
+          IFNULL(m.rankName, '') AS rankName,
+          IFNULL(m.teamName, '') AS teamName,
+          IFNULL(m.retireFlag, '') AS retireFlag,
+          IFNULL(m.isRetired, 0) AS isRetired,
+          IFNULL(d.totalHours, 0) AS totalHours,
+          IFNULL(d.totalRows, 0) AS totalRows,
+          0 AS siteHours,
+          0 AS siteRows
+        FROM member m
+        LEFT JOIN d ON d.MemberNo = m.MemberNo
+        ORDER BY CASE WHEN IFNULL(m.korName,'')='' THEN 1 ELSE 0 END, korName ASC, m.MemberNo ASC
+        '''
+        cur = conn.execute(q_without_site, (start_date, end_date))
     cols = [d[0] for d in cur.description]
     return [dict(zip(cols, row)) for row in cur.fetchall()]
 
@@ -4267,10 +4623,49 @@ class Handler(BaseHTTPRequestHandler):
                 refresh = q.get('refresh', ['0'])[0] in ('1', 'true', 'yes')
                 if refresh:
                     detail = fetch_erp_contract_detail(page_name, project_code, project_name)
-                    run_db_write(lambda conn: replace_erp_contract_detail_cache(conn, detail))
+                    def _write_refresh(conn):
+                        replace_erp_contract_detail_cache(conn, detail)
+                        replace_erp_linked_code_cache(
+                            conn,
+                            page_name,
+                            project_code,
+                            detail.get('projectName', '') or project_name,
+                            detail.get('businessCode', ''),
+                            detail.get('linkedCodes') or {},
+                        )
+                    run_db_write(_write_refresh)
                 with open_db_connection() as conn:
                     ensure_db_schema(conn)
                     cached = get_erp_contract_detail_cache(conn, page_name, project_code)
+                if cached and (
+                    (
+                        not (cached.get('linkedCodes') or {}).get('salesCode')
+                        and not (cached.get('linkedCodes') or {}).get('designCode')
+                    ) or (
+                        (cached.get('linkedCodes') or {}).get('salesCode')
+                        and not (cached.get('linkedCodes') or {}).get('salesName')
+                    ) or (
+                        (cached.get('linkedCodes') or {}).get('designCode')
+                        and not (cached.get('linkedCodes') or {}).get('designName')
+                    )
+                ):
+                    linked_codes = find_erp_linked_codes(project_code, cached.get('businessCode', ''), cached.get('projectName', '') or project_name)
+                    if linked_codes.get('salesCode') or linked_codes.get('designCode'):
+                        cached['linkedCodes'] = linked_codes
+                        def _write_backfill(conn):
+                            replace_erp_contract_detail_cache(conn, cached)
+                            replace_erp_linked_code_cache(
+                                conn,
+                                page_name,
+                                project_code,
+                                cached.get('projectName', '') or project_name,
+                                cached.get('businessCode', ''),
+                                linked_codes,
+                            )
+                        run_db_write(_write_backfill)
+                        with open_db_connection() as conn:
+                            ensure_db_schema(conn)
+                            cached = get_erp_contract_detail_cache(conn, page_name, project_code)
                 if not cached:
                     return self._json(404, {'ok': False, 'error': '계약정보 캐시가 없습니다.'})
                 return self._json(
@@ -4480,9 +4875,16 @@ class Handler(BaseHTTPRequestHandler):
         if parsed.path == '/api/sync-project-aliases':
             try:
                 with sqlite3.connect(DB_PATH) as conn:
-                    alias_rows = load_project_alias_from_erp()
-                    replace_project_alias(conn, alias_rows)
-                return self._json(200, {'ok': True, 'project_alias_source': 'erp', 'project_alias_loaded': len(alias_rows)})
+                    ensure_db_schema(conn)
+                    sync_info = sync_project_alias_caches(conn)
+                    linked_count = rebuild_erp_linked_code_cache(conn, 'const')
+                return self._json(200, {
+                    'ok': True,
+                    'project_alias_source': 'erp',
+                    'project_alias_loaded': len(sync_info['entries']),
+                    'linked_code_cache_loaded': linked_count,
+                    'syncedAt': sync_info['cacheInfo']['syncedAt'],
+                })
             except Exception as e:
                 traceback.print_exc()
                 return self._json(500, {'ok': False, 'error': str(e)})
@@ -4503,7 +4905,11 @@ class Handler(BaseHTTPRequestHandler):
                 q = parse_qs(parsed.query)
                 page_name = q.get('page', ['const'])[0]
                 result = fetch_erp_project_codes(page_name)
-                sync_info = run_db_write(lambda conn: replace_erp_project_code_cache(conn, result['page'], result['rows']))
+                def _write_project_codes(conn):
+                    sync_info = replace_erp_project_code_cache(conn, result['page'], result['rows'])
+                    linked_count = rebuild_erp_linked_code_cache(conn, result['page']) if result['page'] == 'const' else 0
+                    return sync_info, linked_count
+                sync_info, linked_count = run_db_write(_write_project_codes)
                 return self._json(
                     200,
                     {
@@ -4511,6 +4917,7 @@ class Handler(BaseHTTPRequestHandler):
                         'count': sync_info['count'],
                         'page': sync_info['sourcePage'],
                         'syncedAt': sync_info['syncedAt'],
+                        'linkedCodeCacheLoaded': linked_count,
                     },
                 )
             except Exception as e:
@@ -4524,7 +4931,18 @@ class Handler(BaseHTTPRequestHandler):
                 project_code = q.get('projectCode', [''])[0]
                 project_name = q.get('projectName', [''])[0]
                 detail = fetch_erp_contract_detail(page_name, project_code, project_name)
-                sync_info = run_db_write(lambda conn: replace_erp_contract_detail_cache(conn, detail))
+                def _write_contract(conn):
+                    sync_info = replace_erp_contract_detail_cache(conn, detail)
+                    replace_erp_linked_code_cache(
+                        conn,
+                        page_name,
+                        project_code,
+                        detail.get('projectName', '') or project_name,
+                        detail.get('businessCode', ''),
+                        detail.get('linkedCodes') or {},
+                    )
+                    return sync_info
+                sync_info = run_db_write(_write_contract)
                 with open_db_connection() as conn:
                     ensure_db_schema(conn)
                     cached = get_erp_contract_detail_cache(conn, page_name, project_code)
@@ -4588,13 +5006,14 @@ class Handler(BaseHTTPRequestHandler):
                 ensure_db_schema(conn)
                 used = load_from_mysql_into_sqlite(conn)
                 try:
-                    alias_rows = load_project_alias_from_erp()
+                    sync_info = sync_project_alias_caches(conn)
                     used['project_alias_source'] = 'erp'
-                    replace_project_alias(conn, alias_rows)
-                    used['project_alias_loaded'] = len(alias_rows)
+                    used['project_alias_loaded'] = len(sync_info['entries'])
+                    used['linked_code_cache_loaded'] = rebuild_erp_linked_code_cache(conn, 'const')
                 except Exception as e:
                     used['project_alias_source'] = 'erp_failed_keep_existing'
                     used['project_alias_loaded'] = 0
+                    used['linked_code_cache_loaded'] = 0
                     used['project_alias_error'] = str(e)
                 # member rank는 MySQL member.rankCode -> systemconfig 매핑을 우선 사용
                 # (used['member_rank_source'], used['member_rank_loaded']는 load_from_mysql_into_sqlite에서 설정)
@@ -4633,9 +5052,10 @@ if __name__ == '__main__':
         init_db(conn)
         if os.environ.get('STARTUP_MAINTENANCE', '1') not in ('0', 'false', 'False', 'no'):
             try:
-                alias_rows = load_project_alias_from_erp()
-                print(f'Loaded project aliases from ERP: {len(alias_rows)}')
-                replace_project_alias(conn, alias_rows)
+                sync_info = sync_project_alias_caches(conn)
+                print(f"Loaded project aliases from ERP: {len(sync_info['entries'])}")
+                linked_count = rebuild_erp_linked_code_cache(conn, 'const')
+                print(f'Loaded linked code cache: {linked_count}')
             except Exception as e:
                 print(f'Load project aliases from ERP failed, keep existing aliases: {e}')
             # 직급은 MySQL member.rankCode -> systemconfig 매핑 기반으로 /api/load 시 반영