Spaces:

blueellia
/

casting

Runtime error

Upload app.py

6b4bfdb verified 6 months ago

9.98 kB

import io, re, time, requests, gc
import pandas as pd
import streamlit as st
from concurrent.futures import ThreadPoolExecutor, as_completed
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
from openpyxl.utils import get_column_letter

st.set_page_config(page_title="사업장 주소·전화번호 보완", page_icon="📋", layout="centered")
st.markdown("""<style>
#MainMenu,footer,header{visibility:hidden}
.block-container{max-width:660px!important;padding:2rem 1rem!important}
div.stDownloadButton>button{background:#10b981!important;color:#fff!important;border:none!important;border-radius:8px!important;font-weight:600!important;width:100%!important}
</style>""", unsafe_allow_html=True)

FSC_KEY   = "2446da601452d231f1ff77a4c5ac9598896028fb96e8be586bf7f4be58b6231c"
KAKAO_KEY = "e8eef7246a2cd0f3d2d55af8a806e3ef"
NAV_HDR   = {"User-Agent":"Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15","Referer":"https://map.naver.com/","Accept-Language":"ko-KR,ko;q=0.9"}
WORKERS   = 20

def kakao_keyword(name, addr):
    try:
        r = requests.get("https://dapi.kakao.com/v2/local/search/keyword.json",
            headers={"Authorization":f"KakaoAK {KAKAO_KEY}"},
            params={"query":f"{name} {addr[:20]}","size":5}, timeout=5)
        docs = r.json().get("documents",[])
        gu = re.search(r'[\w]+구|[\w]+군', addr)
        gu = gu.group() if gu else ""
        if gu:
            local = [d for d in docs if gu in d.get("address_name","") or gu in d.get("road_address_name","")]
            if local:
                return next((d for d in local if name[:4] in d.get("place_name","")), local[0])
        return next((d for d in docs if name[:4] in d.get("place_name","")), docs[0] if docs else None)
    except: return None

def fsc_info(name, biz_no):
    try:
        r = requests.get(
            "https://apis.data.go.kr/1160100/service/GetCorpBasicInfoService_V2/getCorpOutline_V2",
            params={"serviceKey":FSC_KEY,"pageNo":1,"numOfRows":5,"resultType":"json","corpNm":name}, timeout=5)
        items = r.json().get("response",{}).get("body",{}).get("items",{}).get("item",[])
        if not items: return None
        if isinstance(items, dict): items = [items]
        bc = re.sub(r'\D','',str(biz_no))
        m = next((i for i in items if re.sub(r'\D','',str(i.get("bzno",""))) == bc), items[0])
        return {"phone":m.get("enpTlno",""), "addr":m.get("enpBsadr","") or m.get("enpDtadr","")}
    except: return None

def naver_phone(name, addr):
    clean = re.sub(r'주식회사|유한회사|\(주\)|\(유\)|[(){}\[\]]','',name).strip()
    gu = re.search(r'[\w]+구', addr); gu = gu.group() if gu else ""
    try:
        r = requests.get("https://m.map.naver.com/search2/search.nhn",
            headers=NAV_HDR, params={"query":f"{clean} {gu}","type":"all"}, timeout=8)
        if r.status_code == 429: time.sleep(30); return naver_phone(name, addr)
        for pat in [r'tel:(0\d{1,2}-?\d{3,4}-?\d{4})',r'(?<!\d)(0\d{1,2}-\d{3,4}-\d{4})(?!\d)']:
            f = re.findall(pat, r.text)
            if f: return f[0]
    except: pass
    return ""

def process_one(row_data):
    idx, name, addr, biz_no, is_corp, gu = row_data
    phone=""; full_addr=addr; lat=None; lon=None

    if is_corp:
        info = fsc_info(name, biz_no)
        if info:
            phone = info.get("phone","")
            # FSC 주소는 본사 주소라 덮어쓰지 않음
        time.sleep(0.1)

    kk = kakao_keyword(name, addr)
    if kk:
        if not phone: phone = kk.get("phone","")
        road = kk.get("road_address_name","") or kk.get("address_name","")
        # 카카오 결과가 같은 구/군일 때만 주소 업데이트
        if road and gu and gu in road:
            full_addr = road
            lat = float(kk.get("y") or 0) or None
            lon = float(kk.get("x") or 0) or None
        elif not road or (gu and gu not in road):
            # 엉뚱한 지역 결과 → 원본 주소 유지, 좌표만 취득
            lat = float(kk.get("y") or 0) or None
            lon = float(kk.get("x") or 0) or None

    if not phone: phone = naver_phone(name, full_addr)
    return idx, {"phone":phone, "addr":full_addr, "lat":lat, "lon":lon}

def make_excel(df):
    wb = Workbook(); ws = wb.active; ws.title = "보완완료"
    def thin():
        s = Side(style="thin",color="D0D0D0")
        return Border(left=s,right=s,top=s,bottom=s)
    hdrs = list(df.columns)
    widths = {"사업장명":26,"업종":18,"법인여부":8,"주소":42,"전화번호":14,"가입자수":8,"당월고지금액":14,"위도":12,"경도":12}
    CENTER = {"법인여부","가입자수","당월고지금액","위도","경도","전화번호"}
    for c,h in enumerate(hdrs,1):
        cell = ws.cell(1,c,h)
        cell.fill = PatternFill("solid",start_color="1F3864")
        cell.font = Font(name="Arial",size=10,bold=True,color="FFFFFF")
        cell.alignment = Alignment(horizontal="center",vertical="center")
        cell.border = thin()
        ws.column_dimensions[get_column_letter(c)].width = widths.get(h,12)
    ws.row_dimensions[1].height = 18
    ws.freeze_panes = "A2"
    ws.auto_filter.ref = f"A1:{get_column_letter(len(hdrs))}1"
    for ri,row in enumerate(df.itertuples(index=False),2):
        for ci,(h,v) in enumerate(zip(hdrs,row),1):
            if pd.isna(v) or str(v) in ("nan","None"): v=""
            cell = ws.cell(ri,ci,v)
            cell.font = Font(name="Arial",size=10)
            cell.border = thin()
            cell.fill = PatternFill("solid",start_color="FFFFFF" if ri%2==0 else "F8FAFC")
            cell.alignment = Alignment(horizontal="center" if h in CENTER else "left",vertical="center")
    buf = io.BytesIO(); wb.save(buf); return buf.getvalue()

# ── UI ──────────────────────────────────────────────────
st.title("📋 사업장 주소·전화번호 보완")
st.caption("지역 파일 업로드 → 자동 처리 → 다운로드")
st.divider()

uploaded = st.file_uploader("파일 선택 (국민연금_XX시_XX구_사업장.xlsx)", type=["xlsx"])

if uploaded:
    df = pd.read_excel(uploaded)
    df.columns = [c.strip() for c in df.columns]
    gc.collect()

    name_col = next((c for c in df.columns if "사업장명" in c or "상호" in c), None)
    addr_col = next((c for c in df.columns if "주소" in c or "도로명" in c), None)
    biz_col  = next((c for c in df.columns if "사업자" in c and "번호" in c), None)
    type_col = next((c for c in df.columns if "형태" in c or "법인" in c), None)
    emp_col  = next((c for c in df.columns if "가입자" in c), None)
    pay_col  = next((c for c in df.columns if "고지" in c), None)
    ind_col  = next((c for c in df.columns if "업종" in c), None)

    for col in ["전화번호"]:
        if col not in df.columns: df[col] = ""
        else: df[col] = df[col].fillna("").astype(str).replace("nan","")
    for col in ["위도","경도"]:
        if col not in df.columns: df[col] = None
        else: df[col] = pd.to_numeric(df[col], errors="coerce")

    total = len(df)
    fname = uploaded.name.replace(".xlsx","")
    st.success(f"✅ {fname} — {total:,}개 로드 완료")

    prog   = st.progress(0, text="처리 중...")
    phone_ok = coord_ok = completed = 0

    # 구/군 추출 (엉뚱한 지역 필터링용)
    gu = ""
    if addr_col:
        sample = df[addr_col].dropna().iloc[0] if len(df) > 0 else ""
        m = re.search(r'[\w]+구|[\w]+군', str(sample))
        gu = m.group() if m else ""

    tasks = []
    for idx in df.index:
        row = df.loc[idx]
        tasks.append((idx,
            str(row.get(name_col,"")) if name_col else "",
            str(row.get(addr_col,"")) if addr_col else "",
            str(row.get(biz_col,""))  if biz_col  else "",
            str(row.get(type_col,"")) in ["법인","1"] if type_col else False,
            gu))

    with ThreadPoolExecutor(max_workers=WORKERS) as executor:
        futures = {executor.submit(process_one, t): t[0] for t in tasks}
        for future in as_completed(futures):
            try:
                idx, res = future.result()
                df.loc[idx,"전화번호"] = str(res["phone"])
                df.loc[idx,"위도"]    = res["lat"]
                df.loc[idx,"경도"]    = res["lon"]
                if addr_col and res["addr"]: df.loc[idx,addr_col] = res["addr"]
                if res["phone"]: phone_ok += 1
                if res["lat"]:   coord_ok += 1
            except: pass
            completed += 1
            prog.progress(completed/total, text=f"📞{phone_ok} 📍{coord_ok} | {completed}/{total}")

    prog.empty()

    col_map = {}
    if name_col: col_map[name_col]="사업장명"
    if type_col: col_map[type_col]="법인여부"
    if addr_col: col_map[addr_col]="주소"
    col_map["전화번호"]="전화번호"
    if emp_col:  col_map[emp_col]="가입자수"
    if pay_col:  col_map[pay_col]="당월고지금액"
    if ind_col:  col_map[ind_col]="업종"
    col_map["위도"]="위도"; col_map["경도"]="경도"

    out = df[[c for c in col_map if c in df.columns]].rename(columns=col_map)
    # 전화번호/주소 없으면 삭제 + 해당 구/군 아닌 주소도 삭제
    mask = (out["전화번호"].notna() & (out["전화번호"]!="") &
            out["주소"].notna()     & (out["주소"]!=""))
    if gu:
        mask = mask & out["주소"].str.contains(gu, na=False)
    out = out[mask].reset_index(drop=True)
    del df; gc.collect()

    st.success(f"✅ 완료! 📞{phone_ok:,}개 | 📍{coord_ok:,}개 | 최종 {len(out):,}개")
    st.balloons()
    st.download_button("⬇️ 보완 완료 엑셀 다운로드",
        make_excel(out), fname+"_보완완료.xlsx",
        "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
        use_container_width=True)

	import io, re, time, requests, gc
	import pandas as pd
	import streamlit as st
	from concurrent.futures import ThreadPoolExecutor, as_completed
	from openpyxl import Workbook
	from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
	from openpyxl.utils import get_column_letter

	st.set_page_config(page_title="사업장 주소·전화번호 보완", page_icon="📋", layout="centered")
	st.markdown("""<style>
	#MainMenu,footer,header{visibility:hidden}
	.block-container{max-width:660px!important;padding:2rem 1rem!important}
	div.stDownloadButton>button{background:#10b981!important;color:#fff!important;border:none!important;border-radius:8px!important;font-weight:600!important;width:100%!important}
	</style>""", unsafe_allow_html=True)

	FSC_KEY = "2446da601452d231f1ff77a4c5ac9598896028fb96e8be586bf7f4be58b6231c"
	KAKAO_KEY = "e8eef7246a2cd0f3d2d55af8a806e3ef"
	NAV_HDR = {"User-Agent":"Mozilla/5.0 (iPhone; CPU iPhone OS 16_0 like Mac OS X) AppleWebKit/605.1.15","Referer":"https://map.naver.com/","Accept-Language":"ko-KR,ko;q=0.9"}
	WORKERS = 20

	def kakao_keyword(name, addr):
	try:
	r = requests.get("https://dapi.kakao.com/v2/local/search/keyword.json",
	headers={"Authorization":f"KakaoAK {KAKAO_KEY}"},
	params={"query":f"{name} {addr[:20]}","size":5}, timeout=5)
	docs = r.json().get("documents",[])
	gu = re.search(r'[\w]+구\|[\w]+군', addr)
	gu = gu.group() if gu else ""
	if gu:
	local = [d for d in docs if gu in d.get("address_name","") or gu in d.get("road_address_name","")]
	if local:
	return next((d for d in local if name[:4] in d.get("place_name","")), local[0])
	return next((d for d in docs if name[:4] in d.get("place_name","")), docs[0] if docs else None)
	except: return None

	def fsc_info(name, biz_no):
	try:
	r = requests.get(
	"https://apis.data.go.kr/1160100/service/GetCorpBasicInfoService_V2/getCorpOutline_V2",
	params={"serviceKey":FSC_KEY,"pageNo":1,"numOfRows":5,"resultType":"json","corpNm":name}, timeout=5)
	items = r.json().get("response",{}).get("body",{}).get("items",{}).get("item",[])
	if not items: return None
	if isinstance(items, dict): items = [items]
	bc = re.sub(r'\D','',str(biz_no))
	m = next((i for i in items if re.sub(r'\D','',str(i.get("bzno",""))) == bc), items[0])
	return {"phone":m.get("enpTlno",""), "addr":m.get("enpBsadr","") or m.get("enpDtadr","")}
	except: return None

	def naver_phone(name, addr):
	clean = re.sub(r'주식회사\|유한회사\|\(주\)\|\(유\)\|[(){}\[\]]','',name).strip()
	gu = re.search(r'[\w]+구', addr); gu = gu.group() if gu else ""
	try:
	r = requests.get("https://m.map.naver.com/search2/search.nhn",
	headers=NAV_HDR, params={"query":f"{clean} {gu}","type":"all"}, timeout=8)
	if r.status_code == 429: time.sleep(30); return naver_phone(name, addr)
	for pat in [r'tel:(0\d{1,2}-?\d{3,4}-?\d{4})',r'(?<!\d)(0\d{1,2}-\d{3,4}-\d{4})(?!\d)']:
	f = re.findall(pat, r.text)
	if f: return f[0]
	except: pass
	return ""

	def process_one(row_data):
	idx, name, addr, biz_no, is_corp, gu = row_data
	phone=""; full_addr=addr; lat=None; lon=None

	if is_corp:
	info = fsc_info(name, biz_no)
	if info:
	phone = info.get("phone","")
	# FSC 주소는 본사 주소라 덮어쓰지 않음
	time.sleep(0.1)

	kk = kakao_keyword(name, addr)
	if kk:
	if not phone: phone = kk.get("phone","")
	road = kk.get("road_address_name","") or kk.get("address_name","")
	# 카카오 결과가 같은 구/군일 때만 주소 업데이트
	if road and gu and gu in road:
	full_addr = road
	lat = float(kk.get("y") or 0) or None
	lon = float(kk.get("x") or 0) or None
	elif not road or (gu and gu not in road):
	# 엉뚱한 지역 결과 → 원본 주소 유지, 좌표만 취득
	lat = float(kk.get("y") or 0) or None
	lon = float(kk.get("x") or 0) or None

	if not phone: phone = naver_phone(name, full_addr)
	return idx, {"phone":phone, "addr":full_addr, "lat":lat, "lon":lon}

	def make_excel(df):
	wb = Workbook(); ws = wb.active; ws.title = "보완완료"
	def thin():
	s = Side(style="thin",color="D0D0D0")
	return Border(left=s,right=s,top=s,bottom=s)
	hdrs = list(df.columns)
	widths = {"사업장명":26,"업종":18,"법인여부":8,"주소":42,"전화번호":14,"가입자수":8,"당월고지금액":14,"위도":12,"경도":12}
	CENTER = {"법인여부","가입자수","당월고지금액","위도","경도","전화번호"}
	for c,h in enumerate(hdrs,1):
	cell = ws.cell(1,c,h)
	cell.fill = PatternFill("solid",start_color="1F3864")
	cell.font = Font(name="Arial",size=10,bold=True,color="FFFFFF")
	cell.alignment = Alignment(horizontal="center",vertical="center")
	cell.border = thin()
	ws.column_dimensions[get_column_letter(c)].width = widths.get(h,12)
	ws.row_dimensions[1].height = 18
	ws.freeze_panes = "A2"
	ws.auto_filter.ref = f"A1:{get_column_letter(len(hdrs))}1"
	for ri,row in enumerate(df.itertuples(index=False),2):
	for ci,(h,v) in enumerate(zip(hdrs,row),1):
	if pd.isna(v) or str(v) in ("nan","None"): v=""
	cell = ws.cell(ri,ci,v)
	cell.font = Font(name="Arial",size=10)
	cell.border = thin()
	cell.fill = PatternFill("solid",start_color="FFFFFF" if ri%2==0 else "F8FAFC")
	cell.alignment = Alignment(horizontal="center" if h in CENTER else "left",vertical="center")
	buf = io.BytesIO(); wb.save(buf); return buf.getvalue()

	# ── UI ──────────────────────────────────────────────────
	st.title("📋 사업장 주소·전화번호 보완")
	st.caption("지역 파일 업로드 → 자동 처리 → 다운로드")
	st.divider()

	uploaded = st.file_uploader("파일 선택 (국민연금_XX시_XX구_사업장.xlsx)", type=["xlsx"])

	if uploaded:
	df = pd.read_excel(uploaded)
	df.columns = [c.strip() for c in df.columns]
	gc.collect()

	name_col = next((c for c in df.columns if "사업장명" in c or "상호" in c), None)
	addr_col = next((c for c in df.columns if "주소" in c or "도로명" in c), None)
	biz_col = next((c for c in df.columns if "사업자" in c and "번호" in c), None)
	type_col = next((c for c in df.columns if "형태" in c or "법인" in c), None)
	emp_col = next((c for c in df.columns if "가입자" in c), None)
	pay_col = next((c for c in df.columns if "고지" in c), None)
	ind_col = next((c for c in df.columns if "업종" in c), None)

	for col in ["전화번호"]:
	if col not in df.columns: df[col] = ""
	else: df[col] = df[col].fillna("").astype(str).replace("nan","")
	for col in ["위도","경도"]:
	if col not in df.columns: df[col] = None
	else: df[col] = pd.to_numeric(df[col], errors="coerce")

	total = len(df)
	fname = uploaded.name.replace(".xlsx","")
	st.success(f"✅ {fname} — {total:,}개 로드 완료")

	prog = st.progress(0, text="처리 중...")
	phone_ok = coord_ok = completed = 0

	# 구/군 추출 (엉뚱한 지역 필터링용)
	gu = ""
	if addr_col:
	sample = df[addr_col].dropna().iloc[0] if len(df) > 0 else ""
	m = re.search(r'[\w]+구\|[\w]+군', str(sample))
	gu = m.group() if m else ""

	tasks = []
	for idx in df.index:
	row = df.loc[idx]
	tasks.append((idx,
	str(row.get(name_col,"")) if name_col else "",
	str(row.get(addr_col,"")) if addr_col else "",
	str(row.get(biz_col,"")) if biz_col else "",
	str(row.get(type_col,"")) in ["법인","1"] if type_col else False,
	gu))

	with ThreadPoolExecutor(max_workers=WORKERS) as executor:
	futures = {executor.submit(process_one, t): t[0] for t in tasks}
	for future in as_completed(futures):
	try:
	idx, res = future.result()
	df.loc[idx,"전화번호"] = str(res["phone"])
	df.loc[idx,"위도"] = res["lat"]
	df.loc[idx,"경도"] = res["lon"]
	if addr_col and res["addr"]: df.loc[idx,addr_col] = res["addr"]
	if res["phone"]: phone_ok += 1
	if res["lat"]: coord_ok += 1
	except: pass
	completed += 1
	prog.progress(completed/total, text=f"📞{phone_ok} 📍{coord_ok} \| {completed}/{total}")

	prog.empty()

	col_map = {}
	if name_col: col_map[name_col]="사업장명"
	if type_col: col_map[type_col]="법인여부"
	if addr_col: col_map[addr_col]="주소"
	col_map["전화번호"]="전화번호"
	if emp_col: col_map[emp_col]="가입자수"
	if pay_col: col_map[pay_col]="당월고지금액"
	if ind_col: col_map[ind_col]="업종"
	col_map["위도"]="위도"; col_map["경도"]="경도"

	out = df[[c for c in col_map if c in df.columns]].rename(columns=col_map)
	# 전화번호/주소 없으면 삭제 + 해당 구/군 아닌 주소도 삭제
	mask = (out["전화번호"].notna() & (out["전화번호"]!="") &
	out["주소"].notna() & (out["주소"]!=""))
	if gu:
	mask = mask & out["주소"].str.contains(gu, na=False)
	out = out[mask].reset_index(drop=True)
	del df; gc.collect()

	st.success(f"✅ 완료! 📞{phone_ok:,}개 \| 📍{coord_ok:,}개 \| 최종 {len(out):,}개")
	st.balloons()
	st.download_button("⬇️ 보완 완료 엑셀 다운로드",
	make_excel(out), fname+"_보완완료.xlsx",
	"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
	use_container_width=True)