user01/mapindrag
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feat: 근무처 주소 중복 시 표시/내보내기 통합, 근무자 클릭 시 지도 이동

Browse Source
This commit is contained in:
user01
2026-05-24 23:56:26 +09:00
parent ffca14f2e6
commit e62c10e66f
3 changed files with 247 additions and 217 deletions
Download Patch File Download Diff File Expand all files Collapse all files

275
app.py
View File

@@ -450,8 +450,12 @@ def download_from_google_drive(file_id, timeout=30):
return None
def process_excel_file(wb, filename):
"""엑셀 파일을 처리하고 근무자/근무처 데이터를 반환합니다."""
def process_excel_file(wb, filename, force_type=None):
"""엑셀 파일을 처리하고 근무자/근무처 데이터를 반환합니다.
force_type: 'worker' - 모든 시트를 근무자로 처리
'workplace' - 모든 시트를 근무처로 처리
None - 자동 감지
"""
workers, workplaces, all_addrs = [], [], set()
for sheet_name in wb.sheetnames:
@@ -461,7 +465,12 @@ def process_excel_file(wb, filename):
update_progress('엑셀 파싱', f'시트 처리 중: {sheet_name}')
hope_idx = find_header_idx(headers, ['희망사항', '희망', '배치희망'])
is_worker_sheet = is_likely_worker_sheet(sheet_name, headers)
if force_type == 'worker':
is_worker_sheet = True
elif force_type == 'workplace':
is_worker_sheet = False
else:
is_worker_sheet = is_likely_worker_sheet(sheet_name, headers)
name_idx = find_header_idx(headers, ['이름', '성명', '명칭', '근무지명', '투표소명'])
addr_idx = find_header_idx(headers, ['주소', '위치', '소재지'])
@@ -547,7 +556,12 @@ def process_excel_file(wb, filename):
headers = [str(c.value or '').strip() for c in ws[1]]
hope_idx = find_header_idx(headers, ['희망사항', '희망', '배치희망'])
is_worker_sheet = is_likely_worker_sheet(sheet_name, headers)
if force_type == 'worker':
is_worker_sheet = True
elif force_type == 'workplace':
is_worker_sheet = False
else:
is_worker_sheet = is_likely_worker_sheet(sheet_name, headers)
name_idx = find_header_idx(headers, ['이름', '성명', '명칭', '근무지명', '투표소명'])
addr_idx = find_header_idx(headers, ['주소', '위치', '소재지'])
@@ -596,187 +610,81 @@ def process_excel_file(wb, filename):
return workers, workplaces
@app.route('/api/upload', methods=['POST'])
def upload():
tid = request.form.get('tab') or store['active_tab']
def _handle_upload(tid, file, force_type):
"""공통 업로드 처리 로직"""
t = store['tabs'].get(tid)
if not t:
return jsonify({'error': '탭 없음'}), 400
file = request.files.get('file')
if not file:
return jsonify({'error': '파일이 없습니다.'}), 400
return None, jsonify({'error': '탭 없음'}), 400
upload_progress.update({
'active': True,
'stage': '시작',
'total': 0,
'current': 0,
'message': '',
'logs': [],
'active': True, 'stage': '시작', 'total': 0, 'current': 0, 'message': '', 'logs': [],
})
update_progress('파일 로딩', f'파일을 읽는 중: {file.filename}')
path = os.path.join(app.config['UPLOAD_FOLDER'], file.filename)
file.save(path)
wb = openpyxl.load_workbook(path, data_only=True)
logger.info(f'=== 엑셀 업로드 시작: {file.filename} ===')
logger.info(f'=== {"근무자" if force_type == "worker" else "근무처"} 엑셀 업로드 시작: {file.filename} ===')
logger.info(f'시트 목록: {wb.sheetnames}')
workers, workplaces, all_addrs = [], [], set()
workers, workplaces = process_excel_file(wb, file.filename, force_type=force_type)
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
headers = [str(c.value or '').strip() for c in ws[1]]
logger.info(f' 시트: {sheet_name}, 헤더: {headers}')
update_progress('엑셀 파싱', f'시트 처리 중: {sheet_name}')
hope_idx = find_header_idx(headers, ['희망사항', '희망', '배치희망'])
is_worker_sheet = is_likely_worker_sheet(sheet_name, headers)
name_idx = find_header_idx(headers, ['이름', '성명', '명칭', '근무지명', '투표소명'])
addr_idx = find_header_idx(headers, ['주소', '위치', '소재지'])
logger.info(f' → 희망사항: {hope_idx}, 이름: {name_idx}, 주소: {addr_idx}, 근무자시트: {is_worker_sheet}')
if is_worker_sheet:
hope_idx = hope_idx if hope_idx is not None else 1
name_idx = name_idx if name_idx is not None else 0
dob_idx = find_header_idx(headers, ['생년월일', '생일']) or 3
phone_idx = find_header_idx(headers, ['연락처', '전화', '휴대폰']) or 4
if addr_idx is None:
addr_idx = 2
logger.warning(f' [주의] 주소 컬럼을 찾을 수 없어 기본 인덱스 {addr_idx} 사용')
row_count = 0
for row in ws.iter_rows(min_row=2, values_only=True):
if not any(row): continue
name = str(row[name_idx] or '').strip() if name_idx is not None else ''
if not name: continue
addr = str(row[addr_idx] or '').strip()
if addr:
all_addrs.add(addr)
row_count += 1
logger.info(f' → 근무자 행 수: {row_count}')
else:
name_idx = name_idx if name_idx is not None else 0
if addr_idx is None:
addr_idx = 1
logger.warning(f' [주의] 주소 컬럼을 찾을 수 없어 기본 인덱스 {addr_idx} 사용')
row_count = 0
for row in ws.iter_rows(min_row=2, values_only=True):
if not any(row): continue
name = str(row[name_idx] or '').strip() if name_idx is not None else ''
if not name: continue
addr = str(row[addr_idx] or '').strip()
if addr:
all_addrs.add(addr)
row_count += 1
logger.info(f' → 근무처 행 수: {row_count}')
total_addrs = len(all_addrs)
logger.info(f'수집된 고유 주소 수: {total_addrs}')
GEO_STATS['success'] = 0
GEO_STATS['failed'] = 0
GEO_STATS['failed_addrs'] = []
if VWORLD_API_KEY:
api_name = 'VWorld API'
delay = 0.1
elif KAKAO_API_KEY:
api_name = '카카오 API'
delay = 0.1
else:
api_name = 'Nominatim (API 키 권장)'
logger.info(' (참고: VWorld API 키 발급받으시면 정확도와 속도가 크게 향상됩니다)')
delay = 1.0
logger.info(f'=== 지오코딩 시작 ({api_name}) ===')
update_progress('지오코딩', f'주소를 좌표로 변환 중 ({api_name})', total=total_addrs, current=0)
geo_results = {}
for idx, addr in enumerate(all_addrs, 1):
msg = f'[{idx}/{total_addrs}] {addr}'
logger.info(f' {msg}')
geo_results[addr] = geocode_with_fallback(addr)
if idx % 5 == 0 or idx == total_addrs:
update_progress('지오코딩', f'좌표 변환 중... {idx}/{total_addrs}', total=total_addrs, current=idx)
if idx < total_addrs:
time.sleep(delay)
logger.info(f'=== 지오코딩 완료: 성공 {GEO_STATS["success"]}, 실패 {GEO_STATS["failed"]} ===')
if GEO_STATS['failed_addrs']:
logger.warning(f'실패한 주소 목록 (최대 20개): {GEO_STATS["failed_addrs"]}')
update_progress('완료', f'처리 완료: 근무자 {len(workers or [])}명, 근무처 {len(workplaces or [])}개')
upload_progress['active'] = False
for sheet_name in wb.sheetnames:
ws = wb[sheet_name]
headers = [str(c.value or '').strip() for c in ws[1]]
hope_idx = find_header_idx(headers, ['희망사항', '희망', '배치희망'])
is_worker_sheet = is_likely_worker_sheet(sheet_name, headers)
name_idx = find_header_idx(headers, ['이름', '성명', '명칭', '근무지명', '투표소명'])
addr_idx = find_header_idx(headers, ['주소', '위치', '소재지'])
if is_worker_sheet:
hope_idx = hope_idx if hope_idx is not None else 1
name_idx = name_idx if name_idx is not None else 0
dob_idx = find_header_idx(headers, ['생년월일', '생일']) or 3
phone_idx = find_header_idx(headers, ['연락처', '전화', '휴대폰']) or 4
if addr_idx is None:
addr_idx = 2
for row in ws.iter_rows(min_row=2, values_only=True):
if not any(row): continue
name = str(row[name_idx] or '').strip() if name_idx is not None else ''
if not name: continue
addr = str(row[addr_idx] or '').strip()
lat, lng = geo_results.get(addr, (YONGSAN_CENTER[0] + random.uniform(-0.005, 0.005),
YONGSAN_CENTER[1] + random.uniform(-0.005, 0.005)))
workers.append({
'id': f'w{len(workers)}', 'name': name,
'hope': str(row[hope_idx] or '').strip(),
'address': addr,
'dob': str(row[dob_idx] or '').strip(),
'phone': str(row[phone_idx] or '').strip(),
'lat': lat, 'lng': lng,
})
else:
name_idx = name_idx if name_idx is not None else 0
if addr_idx is None:
addr_idx = 1
for row in ws.iter_rows(min_row=2, values_only=True):
if not any(row): continue
name = str(row[name_idx] or '').strip() if name_idx is not None else ''
if not name: continue
addr = str(row[addr_idx] or '').strip()
lat, lng = geo_results.get(addr, (YONGSAN_CENTER[0] + random.uniform(-0.005, 0.005),
YONGSAN_CENTER[1] + random.uniform(-0.005, 0.005)))
workplaces.append({
'id': f'p{len(workplaces)}', 'name': name,
'address': addr, 'lat': lat, 'lng': lng,
})
logger.info(f'=== 업로드 완료: 근무자 {len(workers)}명, 근무처 {len(workplaces)}개 ===')
t['workers'] = workers
t['workplaces'] = workplaces
t['assignments'] = {}
store['filename'] = file.filename
update_progress('완료', f'처리 완료: 근무자 {len(workers)}명, 근무처 {len(workplaces)}개')
return (workers, workplaces), None, None
@app.route('/api/upload-workers', methods=['POST'])
def upload_workers():
tid = request.form.get('tab') or store['active_tab']
file = request.files.get('file')
if not file:
return jsonify({'error': '파일이 없습니다.'}), 400
result, err, status = _handle_upload(tid, file, force_type='worker')
if err:
return err, status
workers, _ = result
t = store['tabs'].get(tid)
t['workers'] = workers
t['assignments'] = {}
update_progress('완료', f'근무자 업로드 완료: {len(workers)}명')
upload_progress['active'] = False
return jsonify({'workers': workers, 'workplaces': workplaces, 'assignments': {}})
return jsonify({'workers': workers, 'workplaces': t['workplaces'], 'assignments': {}})
@app.route('/api/upload-workplaces', methods=['POST'])
def upload_workplaces():
tid = request.form.get('tab') or store['active_tab']
file = request.files.get('file')
if not file:
return jsonify({'error': '파일이 없습니다.'}), 400
result, err, status = _handle_upload(tid, file, force_type='workplace')
if err:
return err, status
_, new_workplaces = result
t = store['tabs'].get(tid)
# 기존 근무처에 추가 (append) — ID 중복 방지
existing = t['workplaces']
offset = len(existing)
for wp in new_workplaces:
wp['id'] = f'p{offset}'
offset += 1
t['workplaces'] = existing + new_workplaces
total = len(t['workplaces'])
update_progress('완료', f'근무처 업로드 완료 (총 {total}개, 이번 {len(new_workplaces)}개)')
upload_progress['active'] = False
return jsonify({'workers': t['workers'], 'workplaces': t['workplaces'], 'assignments': {}})
@app.route('/api/google-drive-upload', methods=['POST'])
@@ -784,6 +692,7 @@ def google_drive_upload():
"""구글 드라이브에서 엑셀 파일을 다운로드하여 처리합니다."""
body = request.get_json() or {}
file_id = body.get('fileId', '').strip()
upload_type = body.get('type', 'auto') # 'worker', 'workplace', 'auto'
tid = body.get('tab') or store['active_tab']
if not file_id:
@@ -815,24 +724,31 @@ def google_drive_upload():
update_progress('파일 파싱', '엑셀 파일을 읽는 중입니다...')
wb = openpyxl.load_workbook(file_content, data_only=True)
logger.info(f'=== 구글 드라이브 엑셀 업로드 시작: {file_id} ===')
force_type = upload_type if upload_type != 'auto' else None
type_label = {'worker': '근무자', 'workplace': '근무처', 'auto': '자동감지'}.get(upload_type, '자동감지')
logger.info(f'=== 구글 드라이브 엑셀 업로드 시작: {file_id} (유형: {type_label}) ===')
logger.info(f'시트 목록: {wb.sheetnames}')
# 엑셀 파일 처리
workers, workplaces = process_excel_file(wb, f'GoogleDrive_{file_id}.xlsx')
workers, workplaces = process_excel_file(wb, f'GoogleDrive_{file_id}.xlsx', force_type=force_type)
# 탭에 데이터 저장
t['workers'] = workers
t['workplaces'] = workplaces
if upload_type == 'worker':
t['workers'] = workers
elif upload_type == 'workplace':
t['workplaces'] = workplaces
else:
t['workers'] = workers
t['workplaces'] = workplaces
t['assignments'] = {}
store['filename'] = f'GoogleDrive_{file_id}'
update_progress('완료', f'처리 완료: 근무자 {len(workers)}명, 근무처 {len(workplaces)}개')
update_progress('완료', f'처리 완료: 근무자 {len(t["workers"])}명, 근무처 {len(t["workplaces"])}개')
upload_progress['active'] = False
logger.info(f'=== 구글 드라이브 업로드 완료: 근무자 {len(workers)}명, 근무처 {len(workplaces)}개 ===')
logger.info(f'=== 구글 드라이브 업로드 완료: 근무자 {len(t["workers"])}명, 근무처 {len(t["workplaces"])}개 ===')
return jsonify({'workers': workers, 'workplaces': workplaces, 'assignments': {}})
return jsonify({'workers': t['workers'], 'workplaces': t['workplaces'], 'assignments': {}})
except Exception as e:
logger.error(f'엑셀 파일 처리 중 오류: {e}')
@@ -899,9 +815,18 @@ def export():
ws2 = wb.create_sheet('근무지별 배치')
ws2.append(['근무지명', '주소', '배치인원', '배치된근무자'])
# 같은 주소의 근무처는 통합하여 표시
addr_groups = {}
for wp in t['workplaces']:
aw = [w['name'] for w in t['workers'] if t['assignments'].get(w['id']) == wp['id']]
ws2.append([wp['name'], wp['address'], len(aw), ', '.join(aw)])
key = wp['address']
if key not in addr_groups:
addr_groups[key] = {'names': [], 'workplace_ids': []}
addr_groups[key]['names'].append(wp['name'])
addr_groups[key]['workplace_ids'].append(wp['id'])
for addr, g in addr_groups.items():
combined_name = ', '.join(g['names'])
aw = [w['name'] for w in t['workers'] if t['assignments'].get(w['id']) in g['workplace_ids']]
ws2.append([combined_name, addr, len(aw), ', '.join(aw)])
buf = io.BytesIO()
wb.save(buf)