instruction.md
# 장수 점포 레이더 (상가업소 시계열)
## What this skill does
공공데이터포털 **「소상공인시장진흥공단_상가(상권)정보」** 공개 파일(비회원 다운로드 가능)을 이용해 두 가지를 한다.
1. `current` — 최신 분기 스냅샷(전국 17개 시도 zip)에서 **업종코드/상호 키워드로 점포 전수**를 뽑는다. 스냅샷에 존재 = 해당 시점 영업 중으로 취급되는 데이터다.
2. `match` — 사용자가 가진 **과거 스냅샷 CSV**(예: 2019년 국가중점데이터 배포본)와 상호·좌표를 매칭해 **"과거에도 존재했고 지금도 존재하는" 장수 점포**를 추출한다.
기본 업종은 문구·완구(`G21302` 문구/회화용품 소매업, `G21306` 장난감 소매업)이며 옵션으로 어떤 업종이든 지정할 수 있다.
## Honest limitations (반드시 사용자에게 고지)
- 이 데이터에는 **사업자등록번호·개업일이 없다.** 산출되는 것은 "최초 관측 시점 하한"(그 시점에 이미 존재)이지 개업일이 아니다.
- 상호 변경·이전한 점포는 매칭에서 빠진다(과소집계). 동명 상호는 좌표 거리(기본 150m)로 구분하지만 단정하지 않는다.
- 폐업 확정은 사업자등록번호 확보 후 `nts-business-registration`(국세청 상태조회)으로 하고, 인허가 업종이면 `localdata-business-status`로 인허가일자(업력)를 본다.
## Design principles
- 점수·등급 같은 해석 라벨을 만들지 않는다. 스냅샷 존재 사실 + 매칭 방법만 담는다.
- 무인증 공개 파일 서버를 사용자 머신에서 먼저 직접 호출한다. 직접 경로가 timeout/차단되면 GitHub Actions가 검증해 R2에 저장한 공개 미러를 fallback으로 사용한다.
- 미러 ZIP은 `latest.json`의 크기·SHA-256과 ZIP CRC/CSV 존재 검증을 모두 통과해야 캐시로 승격한다. 미러는 API 프록시가 아니라 공개 원본의 검증된 객체 복제본이다.
- 최신 zip은 수백 MB이므로 1일 로컬 캐시(`~/.cache/k-skill/store-longevity-radar/`)한다. 반복 다운로드하지 않는다.
## When to use
- "지금도 영업 중인 오래된 문방구/철물점/레코드점 리스트 뽑아줘"
- "이 업종 전국 점포 전수 데이터 줘 (주소·좌표 포함)"
- "10년 전에도 있었고 아직 있는 가게 찾아줘" (과거 스냅샷 CSV 필요)
## Prerequisites
- 인터넷 연결, `python3` (stdlib만 사용)
- `scripts/store_longevity_radar.py` helper
- `match`를 쓰려면 과거 스냅샷 CSV 파일 (사용자가 보유하거나 별도 확보; 과거분은 공공데이터포털에서 최신 분기만 배포되므로 강의·연구용 미러 등에서 구한다)
## Credential requirements
- 없음. 무인증 공개 파일 다운로드다.
## Inputs
- `--code`: 상권업종소분류코드 (반복 지정, 기본 `G21302` `G21306`). 다른 업종을 `match`할 때 코드체계가 바뀌었다면 현재 코드와 과거 코드를 모두 반복 지정한다.
- `--keyword`: 상호 키워드 (반복 지정, 기본 문구/문방구/완구/장난감 — 업종코드 밖 점포 보완용)
- `--sido`: 시도명 필터 (예: `서울`, `부산`; 생략 시 전국)
- `--zip`: 이미 받아둔 최신 zip 경로 (생략 시 자동 다운로드+캐시)
- `--old-csv`: (`match` 전용) 과거 스냅샷 CSV, 반복 지정. `'|'`/`','` 구분자 자동 감지
- `--max-dist`: (`match` 전용) 동일 상호 허용 좌표 거리(m), 기본 150
- `--out`, `--format`: 출력 파일/형식 (csv 기본, json 가능)
- `KSKILL_STORE_LONGEVITY_MIRROR_MANIFEST_URL`: 검증 미러 manifest override. 기본 `https://pub-c974105a1e4840bcaa264cb2a55d99a1.r2.dev/store-longevity-radar/latest.json`
## CLI examples
```bash
# 전국 문구·완구 현재 전수 → CSV
npx -y @nomadamas/k-skill@0 exec store-longevity-radar scripts/store_longevity_radar.py -- \
current --out 전국_문구완구.csv
# 서울·부산만, 과거 2019 스냅샷과 매칭해 장수 점포 추출
npx -y @nomadamas/k-skill@0 exec store-longevity-radar scripts/store_longevity_radar.py -- \
match --sido 서울 --sido 부산 --old-csv 상가업소정보_201912_01.csv --out 장수점포.csv
# 다른 업종 예: 철물점 상호 키워드로
npx -y @nomadamas/k-skill@0 exec store-longevity-radar scripts/store_longevity_radar.py -- \
current --code NONE --keyword 철물 --out 전국_철물점.csv
```
## Workflow
1. `current`부터 실행해 대상 업종 전수를 확보한다. zip 자동 다운로드는 수 분 걸릴 수 있음을 사용자에게 알린다. 원본 직접 다운로드를 먼저 시도하고 실패하면 크기·SHA-256이 명시된 검증 미러로 자동 전환한다.
2. 과거 스냅샷 CSV가 있으면 `match`로 장수 점포를 추출한다. 기본 문구·완구 코드는 helper가 2022년 이전 `D08A01`/`D04A01`/`D04A02`를 자동 포함한다. 다른 업종은 현재 코드와 과거 코드를 `--code`로 함께 지정한다.
3. 결과 전달 시 위 Honest limitations를 함께 요약한다.
4. 후속 확인이 필요하면 `nts-business-registration`(폐업 확정), `localdata-business-status`(인허가 업력), `kakao-map`(전화번호·현재 등재)을 안내한다.
## Failure modes
- 데이터셋 페이지에서 파일 ID 발견 실패 → `unavailable` + 수동 확인 URL 출력 (분기 개편 시 페이지 구조 변경 가능).
- 공공데이터포털 접속/다운로드 timeout 또는 HTTP 실패 → 검증 미러 fallback.
- 미러 manifest/ZIP 접근 실패, 크기·SHA-256 불일치, ZIP 손상 → 캐시 미승격 + 직접/미러 양쪽 원인을 담은 `unavailable`.
- 다운로드 중단 → `.part` 파일만 남고 캐시로 승격되지 않음. 재실행하면 이어서 새로 받는다.
- `match`에 과거 CSV 미지정 → argparse 에러. 과거분 확보 방법을 사용자에게 안내한다.
- 0건 매칭: 업종코드가 스냅샷 코드체계와 다를 수 있다. `--keyword`만으로 재시도한다.
## Official surfaces
- 데이터셋: <https://www.data.go.kr/data/15083033/fileData.do>
- 다운로드: `https://www.data.go.kr/cmm/cmm/fileDownload.do?atchFileId=<FILE_ID>&fileDetailSn=1` (무인증)
- 검증 미러 manifest: <https://pub-c974105a1e4840bcaa264cb2a55d99a1.r2.dev/store-longevity-radar/latest.json>
- 관련 스킬: `nts-business-registration`, `localdata-business-status`, `kakao-map`
## Done when
- `current` 결과 건수와 출력 파일 경로가 보고되었다.
- `match` 사용 시 매칭 방법(상가업소번호/상호+좌표)별 건수가 보고되었다.
- 개업일이 아닌 "최초 관측 시점 하한"임이 사용자에게 고지되었다.
scripts/store_longevity_download.py
"""Download and verify the latest store-longevity snapshot."""
import hashlib
import json
import os
import re
import sys
import time
import urllib.error
import urllib.request
import zipfile
DATASET_PAGE = "https://www.data.go.kr/data/15083033/fileData.do"
DOWNLOAD_URL = "https://www.data.go.kr/cmm/cmm/fileDownload.do?atchFileId={fid}&fileDetailSn=1"
USER_AGENT = "Mozilla/5.0 (compatible; k-skill-store-longevity-radar)"
CACHE_DIR = os.path.join(os.path.expanduser("~"), ".cache", "k-skill", "store-longevity-radar")
CACHE_TTL = 24 * 3600
MIRROR_MANIFEST_URL = os.environ.get(
"KSKILL_STORE_LONGEVITY_MIRROR_MANIFEST_URL",
"https://pub-c974105a1e4840bcaa264cb2a55d99a1.r2.dev/store-longevity-radar/latest.json",
)
def log(message):
print(message, file=sys.stderr, flush=True)
def http_get(url, timeout=60):
req = urllib.request.Request(
url,
headers={"User-Agent": USER_AGENT, "Accept": "*/*"},
)
return urllib.request.urlopen(req, timeout=timeout)
def discover_atch_file_id():
html = http_get(DATASET_PAGE).read().decode("utf-8", "replace")
ids = re.findall(r"FILE_[0-9]{6,}", html)
if not ids:
raise ValueError(f"현재 파일 ID를 찾지 못함. 수동 확인: {DATASET_PAGE}")
return ids[0]
def download_to_path(url, path, timeout):
with http_get(url, timeout=timeout) as response, open(path, "wb") as output:
while chunk := response.read(1 << 20):
output.write(chunk)
def validate_snapshot_zip(path):
with zipfile.ZipFile(path) as archive:
corrupt_member = archive.testzip()
if corrupt_member is not None:
raise ValueError(f"손상된 ZIP 항목: {corrupt_member}")
if not any(info.filename.lower().endswith(".csv") for info in archive.infolist()):
raise ValueError("스냅샷 ZIP에 CSV가 없음")
def sha256_path(path):
digest = hashlib.sha256()
with open(path, "rb") as source:
while chunk := source.read(1 << 20):
digest.update(chunk)
return digest.hexdigest()
def download_mirror_zip(path):
with http_get(MIRROR_MANIFEST_URL, timeout=30) as response:
manifest = json.load(response)
if not isinstance(manifest, dict):
raise TypeError("미러 manifest가 JSON object가 아님")
object_url = manifest.get("object_url")
expected_size = manifest.get("size_bytes")
expected_sha256 = manifest.get("sha256")
if (
not isinstance(object_url, str)
or not object_url.startswith("https://")
or not isinstance(expected_size, int)
or expected_size <= 0
or not isinstance(expected_sha256, str)
or not re.fullmatch(r"[0-9a-f]{64}", expected_sha256)
):
raise ValueError("미러 manifest 형식이 올바르지 않음")
tmp = path + ".mirror.part"
download_to_path(object_url, tmp, timeout=1800)
if os.path.getsize(tmp) != expected_size:
raise ValueError("미러 ZIP 크기 불일치")
if sha256_path(tmp) != expected_sha256:
raise ValueError("미러 ZIP SHA-256 불일치")
validate_snapshot_zip(tmp)
os.replace(tmp, path)
log(f"mirror cache: {object_url}")
return path
def download_latest_zip():
os.makedirs(CACHE_DIR, exist_ok=True)
path = os.path.join(CACHE_DIR, "sdc_latest.zip")
if os.path.exists(path) and time.time() - os.path.getmtime(path) < CACHE_TTL:
log(f"cache hit: {path}")
return path
try:
fid = discover_atch_file_id()
log(f"downloading {fid} (수백 MB, 수 분 소요) ...")
tmp = path + ".part"
download_to_path(DOWNLOAD_URL.format(fid=fid), tmp, timeout=1800)
validate_snapshot_zip(tmp)
os.replace(tmp, path)
except (
TimeoutError,
urllib.error.URLError,
OSError,
ValueError,
zipfile.BadZipFile,
) as direct_exc:
log(f"direct download unavailable: {direct_exc}; trying verified mirror")
try:
return download_mirror_zip(path)
except (
TimeoutError,
urllib.error.URLError,
OSError,
TypeError,
ValueError,
json.JSONDecodeError,
zipfile.BadZipFile,
) as mirror_exc:
note = (
f"공공데이터포털 직접 다운로드 실패: {direct_exc}. "
f"검증 미러 다운로드 실패: {mirror_exc}. 수동 확인: {DATASET_PAGE}"
)
raise SystemExit(json.dumps({
"status": "unavailable",
"note": note,
}, ensure_ascii=False)) from None
return path
scripts/store_longevity_radar.py
# -*- coding: utf-8 -*-
"""store-longevity-radar — 상가(상권)정보 스냅샷 기반 업종 전수 추출 + 장수 점포 매칭.
python3 stdlib만 사용한다.
subcommands:
current 최신 공개 스냅샷(무인증 zip)에서 업종코드/상호 키워드로 점포 전수 추출
match current 결과에 과거 스냅샷 CSV를 매칭해 장수 점포 추출
데이터 출처: 공공데이터포털 「소상공인시장진흥공단_상가(상권)정보」 파일데이터
(https://www.data.go.kr/data/15083033/fileData.do) — 비회원 다운로드 가능.
"""
import argparse
import csv
import io
import json
import math
import re
import sys
import zipfile
from store_longevity_download import download_latest_zip
DEFAULT_CODES = ["G21302", "G21306"] # 문구/회화용품 소매업, 장난감 소매업 (2022~ 코드체계)
DEFAULT_KEYWORDS = ["문구", "문방구", "완구", "장난감"]
LEGACY_CODES_BY_CURRENT = {
"G21302": {"D08A01"},
"G21306": {"D04A01", "D04A02"},
}
COLS = ["상가업소번호", "상호명", "상권업종소분류코드", "상권업종소분류명",
"시도명", "시군구명", "행정동명", "도로명주소", "지번주소", "경도", "위도"]
def log(msg):
print(msg, file=sys.stderr)
def norm_name(s):
s = re.sub(r"[\s()\[\]{}\-_.·&']", "", s or "")
return re.sub(r"(점|본점|지점)$", "", s)
def row_matches(d, codes, keywords):
if d.get("상권업종소분류코드") in codes:
return True
name = d.get("상호명", "")
return any(k in name for k in keywords)
def historical_codes(codes):
expanded = set(codes)
for code in codes:
expanded.update(LEGACY_CODES_BY_CURRENT.get(code, ()))
return expanded
def iter_csv(fobj, delimiter):
reader = csv.reader(io.TextIOWrapper(fobj, encoding="utf-8", errors="replace"),
delimiter=delimiter)
header = next(reader)
idx = {h: i for i, h in enumerate(header)}
for row in reader:
yield {h: (row[i] if i < len(row) else "") for h, i in idx.items() if h in COLS}
def extract_current(args):
path = args.zip or download_latest_zip()
rows = []
with zipfile.ZipFile(path) as z:
for info in z.infolist():
if not info.filename.endswith(".csv"):
continue
if args.sido and not any(s in info.filename for s in args.sido):
continue
with z.open(info) as f:
for d in iter_csv(f, ","):
if row_matches(d, set(args.code), args.keyword):
rows.append(d)
return rows
def load_old(paths, codes, keywords):
rows = []
for p in paths:
with open(p, "rb") as f:
head = f.read(4096).decode("utf-8", "replace")
delim = "|" if head.count("|") > head.count(",") else ","
with open(p, "rb") as f:
for d in iter_csv(f, delim):
if row_matches(d, codes, keywords):
rows.append(d)
return rows
def dist_m(a, b):
try:
return math.hypot((float(a["경도"]) - float(b["경도"])) * 88800,
(float(a["위도"]) - float(b["위도"])) * 111000)
except (ValueError, KeyError):
return float("inf")
def match_longevity(current, old, max_dist):
old_by_id = {o["상가업소번호"]: o for o in old if o.get("상가업소번호")}
old_by_name = {}
for o in old:
old_by_name.setdefault(norm_name(o.get("상호명")), []).append(o)
out = []
for c in current:
o, how = old_by_id.get(c.get("상가업소번호")), "상가업소번호"
if not o:
best = None
for cand in old_by_name.get(norm_name(c.get("상호명")), []):
d = dist_m(cand, c)
if d <= max_dist and (best is None or d < best[1]):
best = (cand, d)
if best:
o, how = best[0], f"상호+좌표({best[1]:.0f}m)"
if o:
rec = dict(c)
rec["과거상호"] = o.get("상호명", "")
rec["과거업종"] = o.get("상권업종소분류명", "")
rec["매칭방법"] = how
out.append(rec)
return out
def write_out(rows, out, fmt):
if fmt == "json" or not out:
payload = json.dumps({"count": len(rows), "rows": rows}, ensure_ascii=False)
if out:
with open(out, "w", encoding="utf-8") as f:
f.write(payload)
else:
print(payload)
return
cols = list(rows[0].keys()) if rows else COLS
with open(out, "w", encoding="utf-8-sig", newline="") as f:
w = csv.DictWriter(f, fieldnames=cols, extrasaction="ignore")
w.writeheader()
w.writerows(rows)
print(json.dumps({"count": len(rows), "saved": out}, ensure_ascii=False))
def main():
p = argparse.ArgumentParser(description=__doc__)
sub = p.add_subparsers(dest="cmd", required=True)
common = argparse.ArgumentParser(add_help=False)
common.add_argument("--code", action="append", default=None,
help="상권업종소분류코드 (반복 지정, 기본 G21302/G21306)")
common.add_argument("--keyword", action="append", default=None,
help="상호 키워드 (반복 지정, 기본 문구/문방구/완구/장난감)")
common.add_argument("--sido", action="append", default=None,
help="시도명 부분 문자열 필터 (예: 서울, 부산). 생략 시 전국")
common.add_argument("--zip", help="이미 받아둔 최신 스냅샷 zip 경로 (생략 시 자동 다운로드+캐시)")
common.add_argument("--out", help="출력 파일 경로 (생략 시 stdout JSON)")
common.add_argument("--format", choices=["csv", "json"], default="csv")
sub.add_parser("current", parents=[common], help="최신 스냅샷 업종 전수 추출")
mp = sub.add_parser("match", parents=[common], help="과거 스냅샷과 매칭해 장수 점포 추출")
mp.add_argument("--old-csv", action="append", required=True,
help="과거 상가업소 CSV 경로 (반복 지정 가능, '|'/',' 자동 감지)")
mp.add_argument("--max-dist", type=float, default=150.0,
help="상호 동일 시 허용 좌표 거리(m), 기본 150")
args = p.parse_args()
args.code = args.code or DEFAULT_CODES
args.keyword = args.keyword or DEFAULT_KEYWORDS
current = extract_current(args)
log(f"current: {len(current)}건")
if args.cmd == "current":
write_out(current, args.out, args.format)
return
old = load_old(args.old_csv, historical_codes(set(args.code)), args.keyword)
log(f"old: {len(old)}건")
write_out(match_longevity(current, old, args.max_dist), args.out, args.format)
if __name__ == "__main__":
main()
skill.json
{
"name": "store-longevity-radar",
"description": "소상공인시장진흥공단 상가(상권)정보 공개파일(무인증)로 업종·상호 키워드에 맞는 전국 점포 전수를 뽑고, 과거 스냅샷과 상호+좌표 매칭해 'N년 전에도 존재했고 지금도 영업 중'인 장수 점포 리스트를 추출한다. 사업자등록번호 없이 상호 기준이며 개업일이 아닌 최초 관측 시점 하한을 제공한다.",
"profiles": [
"lookup"
],
"frontmatter": "name: store-longevity-radar\ndescription: 소상공인시장진흥공단 상가(상권)정보 공개파일(무인증)로 업종·상호 키워드에 맞는 전국 점포 전수를 뽑고, 과거 스냅샷과 상호+좌표 매칭해 'N년 전에도 존재했고 지금도 영업 중'인 장수 점포 리스트를 추출한다. 사업자등록번호 없이 상호 기준이며 개업일이 아닌 최초 관측 시점 하한을 제공한다.\nlicense: MIT\nmetadata:\n category: business\n locale: ko-KR\n phase: v1"
}
SKILL.md
---
name: store-longevity-radar
description: 소상공인시장진흥공단 상가(상권)정보 공개파일(무인증)로 업종·상호 키워드에 맞는 전국 점포 전수를 뽑고, 과거 스냅샷과 상호+좌표 매칭해 'N년 전에도 존재했고 지금도 영업 중'인 장수 점포 리스트를 추출한다. 사업자등록번호 없이 상호 기준이며 개업일이 아닌 최초 관측 시점 하한을 제공한다.
license: MIT
metadata:
category: business
locale: ko-KR
phase: v1
---
# store-longevity-radar
<!-- k-skill:cli-stub — generated by scripts/generate-skill-stubs.js; edit skill.json / instruction.md instead -->
## Get the full instructions (required first step)
Run this and follow its output as the primary instructions for this skill:
```bash
npx -y @nomadamas/k-skill@0 instruct store-longevity-radar
```
The CLI detects the current runtime (Dolshoi vault/CloakBrowser vs generic) and prints only the applicable instructions, always up to date. Helper files bundled with the CLI are listed by:
```bash
npx -y @nomadamas/k-skill@0 files store-longevity-radar
```
If `npx` is unavailable, install Node.js 18+ or follow https://github.com/NomaDamas/k-skill#readme, or read the source instructions at https://github.com/NomaDamas/k-skill/blob/main/store-longevity-radar/instruction.md.
## Hard rules even without the CLI
- Never execute payment, message/email delivery, final submission, cancellation, or public posting without the user's explicit approval immediately beforehand.
- Never ask for, print, or store plaintext credentials in chat, files, or shell arguments.
- Never bypass legal, physical-presence, CAPTCHA, identity-proofing, or electronic-signature boundaries.
tests/test_store_longevity_radar.py
import hashlib
import io
import json
import sys
import tempfile
import unittest
import zipfile
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts"))
import store_longevity_download as download
import store_longevity_radar as radar
class HistoricalCodesTest(unittest.TestCase):
def test_default_codes_expand_to_matching_legacy_codes(self):
self.assertEqual(
radar.historical_codes({"G21302", "G21306"}),
{"G21302", "G21306", "D08A01", "D04A01", "D04A02"},
)
def test_custom_code_does_not_include_default_legacy_codes(self):
self.assertEqual(radar.historical_codes({"G99999"}), {"G99999"})
class DownloadFailureTest(unittest.TestCase):
def test_timeout_reports_unavailable_json(self):
with (
tempfile.TemporaryDirectory() as cache_dir,
patch.object(download, "CACHE_DIR", cache_dir),
patch.object(download, "http_get", side_effect=TimeoutError("timed out")),
self.assertRaises(SystemExit) as raised,
):
download.download_latest_zip()
payload = json.loads(str(raised.exception))
self.assertEqual(payload["status"], "unavailable")
self.assertIn("timed out", payload["note"])
def test_direct_timeout_falls_back_to_verified_mirror(self):
zip_buffer = io.BytesIO()
with zipfile.ZipFile(zip_buffer, "w") as archive:
archive.writestr(
"상가정보_서울.csv",
"상가업소번호,상호명,상권업종소분류코드\nS001,행복문구,G21302\n",
)
zip_bytes = zip_buffer.getvalue()
object_url = (
"https://pub-c974105a1e4840bcaa264cb2a55d99a1.r2.dev/"
"store-longevity-radar/objects/FILE_1.zip"
)
manifest = json.dumps({
"source_file_id": "FILE_1",
"size_bytes": len(zip_bytes),
"sha256": hashlib.sha256(zip_bytes).hexdigest(),
"object_url": object_url,
}).encode()
def fake_http_get(url, timeout=60):
if url == download.DATASET_PAGE:
raise TimeoutError("timed out")
if url == download.MIRROR_MANIFEST_URL:
return io.BytesIO(manifest)
if url == object_url:
return io.BytesIO(zip_bytes)
raise AssertionError(f"unexpected URL: {url}")
with (
tempfile.TemporaryDirectory() as cache_dir,
patch.object(download, "CACHE_DIR", cache_dir),
patch.object(download, "http_get", side_effect=fake_http_get),
):
path = download.download_latest_zip()
self.assertEqual(Path(path).read_bytes(), zip_bytes)
if __name__ == "__main__":
unittest.main()