#!/usr/bin/env python3 """ 팜큐 약국 관리 시스템 - Flask 애플리케이션 Headscale + Headplane 고도화 관리자 페이지 """ from flask import Flask, render_template, jsonify, request, redirect, url_for import os import json from datetime import datetime import uuid from config import config from utils.database_new import ( init_databases, get_farmq_session, get_dashboard_stats, get_all_pharmacies_with_stats, get_all_machines_with_details, get_machine_detail, get_pharmacy_detail, get_active_alerts, sync_machines_from_headscale, sync_users_from_headscale ) from models.farmq_models import PharmacyInfo from sqlalchemy import or_ import subprocess from utils.proxmox_client import ProxmoxClient def create_app(config_name=None): """Flask 애플리케이션 팩토리""" app = Flask(__name__) # 설정 로드 config_name = config_name or os.environ.get('FLASK_ENV', 'default') app.config.from_object(config[config_name]) # 데이터베이스 초기화 init_databases( headscale_db_uri='sqlite:////srv/headscale-setup/data/db.sqlite', farmq_db_uri='sqlite:///farmq.db' ) # 데이터 동기화 실행 sync_users_from_headscale() sync_machines_from_headscale() # VNC 세션 관리 (메모리 기반) vnc_sessions = {} # Proxmox 서버 설정 PROXMOX_HOST = "pve7.0bin.in" PROXMOX_USERNAME = "root@pam" PROXMOX_PASSWORD = "trajet6640" # 메인 대시보드 @app.route('/') def dashboard(): """메인 대시보드""" try: # 새로운 통합 통계 함수 사용 stats = get_dashboard_stats() stats['alerts'] = get_active_alerts()[:5] # 최신 5개만 stats['performance'] = {'status': 'good', 'summary': '모든 시스템이 정상 작동 중입니다.'} # 약국별 상태 (상위 10개) pharmacies = get_all_pharmacies_with_stats()[:10] return render_template('dashboard/index.html', stats=stats, pharmacies=pharmacies) except Exception as e: print(f"❌ Dashboard error: {e}") return render_template('error.html', error=str(e)), 500 # 약국 관리 @app.route('/pharmacy') def pharmacy_list(): """약국 목록""" try: pharmacies = get_all_pharmacies_with_stats() return render_template('pharmacy/list.html', pharmacies=pharmacies) except Exception as e: return render_template('error.html', error=str(e)), 500 @app.route('/pharmacy/') def pharmacy_detail(pharmacy_id): """약국 상세 정보""" try: detail_data = get_pharmacy_detail(pharmacy_id) if not detail_data: return render_template('error.html', error='약국을 찾을 수 없습니다.'), 404 return render_template('pharmacy/detail.html', pharmacy=detail_data['pharmacy'], machines=detail_data['machines']) except Exception as e: print(f"❌ Pharmacy detail error: {e}") return render_template('error.html', error=str(e)), 500 # 머신 관리 @app.route('/machines') def machine_list(): """머신 목록""" try: machines = get_all_machines_with_details() return render_template('machines/list.html', machines=machines) except Exception as e: print(f"❌ Machine list error: {e}") return render_template('error.html', error=str(e)), 500 @app.route('/machines/') def machine_detail(machine_id): """머신 상세 정보""" try: print(f"🔍 Getting details for machine ID: {machine_id}") details = get_machine_detail(machine_id) if not details: print(f"❌ No details found for machine ID: {machine_id}") return render_template('error.html', error='머신을 찾을 수 없습니다.'), 404 hostname = details.get('hostname', 'Unknown') print(f"✅ Rendering detail page for machine: {hostname}") # 템플릿에 필요한 추가 변수들 is_online = details.get('online', False) last_seen = details.get('last_seen') # 시간 형식화 if last_seen: from datetime import datetime, timezone if isinstance(last_seen, datetime): now = datetime.now(timezone.utc) if last_seen.tzinfo else datetime.now() delta = now - last_seen if delta.days > 0: last_seen_humanized = f"{delta.days}일 전" elif delta.seconds > 3600: last_seen_humanized = f"{delta.seconds // 3600}시간 전" elif delta.seconds > 60: last_seen_humanized = f"{delta.seconds // 60}분 전" else: last_seen_humanized = "방금 전" else: last_seen_humanized = "알 수 없음" else: last_seen_humanized = "알 수 없음" return render_template('machines/detail.html', machine=details, is_online=is_online, last_seen_humanized=last_seen_humanized) except Exception as e: print(f"❌ Error in machine_detail route: {e}") import traceback traceback.print_exc() return render_template('error.html', error=f'머신 상세 정보 로드 중 오류: {str(e)}'), 500 # API 엔드포인트 @app.route('/api/dashboard/stats') def api_dashboard_stats(): """대시보드 통계 API""" try: stats = get_dashboard_stats() stats['performance'] = {'status': 'good', 'summary': '모든 시스템이 정상 작동 중입니다.'} return jsonify(stats) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/alerts') def api_alerts(): """실시간 알림 API""" try: alerts = get_active_alerts() return jsonify(alerts) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/machines//monitoring') def api_machine_monitoring(machine_id): """머신 모니터링 데이터 API""" try: details = get_machine_detail(machine_id) if not details: return jsonify({'error': '머신을 찾을 수 없습니다.'}), 404 # 최근 모니터링 데이터 반환 metrics_history = details.get('metrics_history', []) return jsonify(metrics_history[:20]) # 최근 20개 데이터 except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/sync/machines') def api_sync_machines(): """Headscale에서 머신 정보 동기화 API""" try: result = sync_machines_from_headscale() return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/sync/users') def api_sync_users(): """Headscale에서 사용자 정보 동기화 API""" try: result = sync_users_from_headscale() return jsonify(result) except Exception as e: return jsonify({'error': str(e)}), 500 @app.route('/api/pharmacy/', methods=['GET']) def api_get_pharmacy(pharmacy_id): """개별 약국 정보 조회 API""" try: from utils.database_new import get_farmq_session from models.farmq_models import PharmacyInfo session = get_farmq_session() try: pharmacy = session.query(PharmacyInfo).filter( PharmacyInfo.id == pharmacy_id ).first() if not pharmacy: return jsonify({'error': '약국을 찾을 수 없습니다.'}), 404 return jsonify({ 'pharmacy': { 'id': pharmacy.id, 'pharmacy_name': pharmacy.pharmacy_name or '', 'business_number': pharmacy.business_number or '', 'manager_name': pharmacy.manager_name or '', 'phone': pharmacy.phone or '', 'address': pharmacy.address or '', 'proxmox_host': pharmacy.proxmox_host or '', 'user_id': pharmacy.headscale_user_name or '', # user_id 대신 headscale_user_name 사용 'headscale_user_name': pharmacy.headscale_user_name or '' } }) finally: session.close() except Exception as e: return jsonify({'error': str(e)}), 500 # 약국 관리 API @app.route('/api/pharmacy', methods=['POST']) def api_create_pharmacy(): """새 약국 생성""" try: data = request.get_json() # 필수 필드 확인 pharmacy_name = data.get('pharmacy_name', '').strip() if not pharmacy_name: return jsonify({ 'success': False, 'error': '약국명은 필수입니다.' }), 400 # FARMQ 데이터베이스에 약국 생성 farmq_session = get_farmq_session() try: new_pharmacy = PharmacyInfo( pharmacy_name=pharmacy_name, business_number=data.get('business_number', '').strip(), manager_name=data.get('manager_name', '').strip(), phone=data.get('phone', '').strip(), address=data.get('address', '').strip(), proxmox_host=data.get('proxmox_host', '').strip(), headscale_user_name=data.get('headscale_user_name', '').strip(), status='active' ) farmq_session.add(new_pharmacy) farmq_session.commit() return jsonify({ 'success': True, 'message': f'약국 "{pharmacy_name}"가 성공적으로 생성되었습니다.', 'pharmacy': new_pharmacy.to_dict() }) finally: farmq_session.close() except Exception as e: print(f"❌ 약국 생성 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 @app.route('/api/pharmacy/', methods=['PUT']) def api_update_pharmacy(pharmacy_id): """약국 정보 수정""" try: data = request.get_json() farmq_session = get_farmq_session() try: pharmacy = farmq_session.query(PharmacyInfo).filter( PharmacyInfo.id == pharmacy_id ).first() if not pharmacy: return jsonify({ 'success': False, 'error': '약국을 찾을 수 없습니다.' }), 404 # 필드 업데이트 if 'pharmacy_name' in data: pharmacy.pharmacy_name = data['pharmacy_name'].strip() if 'business_number' in data: pharmacy.business_number = data['business_number'].strip() if 'manager_name' in data: pharmacy.manager_name = data['manager_name'].strip() if 'phone' in data: pharmacy.phone = data['phone'].strip() if 'address' in data: pharmacy.address = data['address'].strip() if 'proxmox_host' in data: pharmacy.proxmox_host = data['proxmox_host'].strip() if 'headscale_user_name' in data: pharmacy.headscale_user_name = data['headscale_user_name'].strip() pharmacy.updated_at = datetime.now() farmq_session.commit() return jsonify({ 'success': True, 'message': f'약국 "{pharmacy.pharmacy_name}" 정보가 수정되었습니다.', 'pharmacy': pharmacy.to_dict() }) finally: farmq_session.close() except Exception as e: print(f"❌ 약국 수정 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 @app.route('/api/pharmacy//delete', methods=['DELETE']) def api_delete_pharmacy(pharmacy_id): """약국 삭제""" try: farmq_session = get_farmq_session() try: pharmacy = farmq_session.query(PharmacyInfo).filter( PharmacyInfo.id == pharmacy_id ).first() if not pharmacy: return jsonify({ 'success': False, 'error': '약국을 찾을 수 없습니다.' }), 404 pharmacy_name = pharmacy.pharmacy_name farmq_session.delete(pharmacy) farmq_session.commit() return jsonify({ 'success': True, 'message': f'약국 "{pharmacy_name}"가 삭제되었습니다.' }) finally: farmq_session.close() except Exception as e: print(f"❌ 약국 삭제 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 @app.route('/api/pharmacies/available', methods=['GET']) def api_get_available_pharmacies(): """매핑되지 않은 약국 목록 가져오기""" try: farmq_session = get_farmq_session() try: # headscale_user_name이 NULL이거나 빈 문자열인 약국들만 가져오기 pharmacies = farmq_session.query(PharmacyInfo).filter( or_( PharmacyInfo.headscale_user_name.is_(None), PharmacyInfo.headscale_user_name == '' ) ).all() pharmacy_list = [] for pharmacy in pharmacies: pharmacy_list.append({ 'id': pharmacy.id, 'pharmacy_name': pharmacy.pharmacy_name, 'manager_name': pharmacy.manager_name or '미등록', 'business_number': pharmacy.business_number, 'address': pharmacy.address }) return jsonify({ 'success': True, 'pharmacies': pharmacy_list, 'count': len(pharmacy_list) }) finally: farmq_session.close() except Exception as e: print(f"❌ 매핑 가능한 약국 목록 조회 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 @app.route('/api/users//link-pharmacy', methods=['POST']) def api_link_user_pharmacy(user_name): """사용자와 약국 연결""" try: data = request.get_json() pharmacy_id = data.get('pharmacy_id') if not pharmacy_id: return jsonify({ 'success': False, 'error': '약국 ID가 필요합니다.' }), 400 farmq_session = get_farmq_session() try: # 약국 존재 확인 pharmacy = farmq_session.query(PharmacyInfo).filter( PharmacyInfo.id == pharmacy_id ).first() if not pharmacy: return jsonify({ 'success': False, 'error': '약국을 찾을 수 없습니다.' }), 404 # 사용자 존재 확인 (Headscale CLI 사용) result = subprocess.run( ['docker', 'exec', 'headscale', 'headscale', 'users', 'list', '-o', 'json'], capture_output=True, text=True, check=True ) users_data = json.loads(result.stdout) user = next((u for u in users_data if u['name'] == user_name), None) if not user: return jsonify({ 'success': False, 'error': f'사용자 "{user_name}"를 찾을 수 없습니다.' }), 404 # 약국에 사용자 연결 pharmacy.headscale_user_name = user_name pharmacy.headscale_user_id = user['id'] pharmacy.updated_at = datetime.now() farmq_session.commit() return jsonify({ 'success': True, 'message': f'사용자 "{user_name}"가 약국 "{pharmacy.pharmacy_name}"에 연결되었습니다.' }) finally: farmq_session.close() except subprocess.CalledProcessError as e: error_msg = e.stderr if e.stderr else e.stdout return jsonify({ 'success': False, 'error': f'사용자 확인 실패: {error_msg}' }), 400 except Exception as e: print(f"❌ 사용자-약국 연결 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 # VNC 관리 라우트들 @app.route('/vms') def vm_list(): """VM 목록 페이지""" try: # Proxmox 클라이언트 생성 및 로그인 client = ProxmoxClient(PROXMOX_HOST, PROXMOX_USERNAME, PROXMOX_PASSWORD) if not client.login(): return render_template('error.html', error='Proxmox 서버에 연결할 수 없습니다.'), 500 # VM 목록 가져오기 vms = client.get_vm_list() # 통계 계산 total_vms = len(vms) running_vms = len([vm for vm in vms if vm.get('status') == 'running']) stopped_vms = total_vms - running_vms vnc_ready_vms = running_vms # 실행 중인 VM은 모두 VNC 가능 return render_template('vm_list.html', vms=vms, host=PROXMOX_HOST, total_vms=total_vms, running_vms=running_vms, stopped_vms=stopped_vms, vnc_ready_vms=vnc_ready_vms) except Exception as e: print(f"❌ VM 목록 오류: {e}") return render_template('error.html', error=str(e)), 500 @app.route('/api/vm/vnc', methods=['POST']) def api_vm_vnc(): """VNC 연결 세션 생성 API""" try: data = request.get_json() node = data.get('node') vmid = int(data.get('vmid')) vm_name = data.get('vm_name', f'VM-{vmid}') # Proxmox 클라이언트 생성 및 로그인 client = ProxmoxClient(PROXMOX_HOST, PROXMOX_USERNAME, PROXMOX_PASSWORD) if not client.login(): return jsonify({'error': 'Proxmox 서버 로그인 실패'}), 500 # VM 상태 확인 vm_status = client.get_vm_status(node, vmid) print(f"🔍 VM {vmid} 상태: {vm_status}") # VNC 티켓 생성 vnc_data = client.get_vnc_ticket(node, vmid) if not vnc_data: return jsonify({'error': 'VNC 티켓 생성 실패'}), 500 # 세션 ID 생성 session_id = str(uuid.uuid4()) # VNC 세션 저장 vnc_sessions[session_id] = { 'node': node, 'vmid': vmid, 'vm_name': vm_name, 'websocket_url': vnc_data['websocket_url'], 'password': vnc_data.get('password', ''), # VNC 패스워드 추가 'vm_status': vm_status.get('status', 'unknown'), # VM 상태 추가 'created_at': datetime.now() } return jsonify({ 'session_id': session_id, 'vm_name': vm_name, 'success': True }) except Exception as e: print(f"❌ VNC API 오류: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/vnc/refresh/', methods=['POST']) def api_vnc_refresh_ticket(session_id): """VNC 티켓 새로고침 API""" try: # 세션 확인 if session_id not in vnc_sessions: return jsonify({'error': '유효하지 않은 VNC 세션입니다.'}), 404 session_data = vnc_sessions[session_id] node = session_data['node'] vmid = session_data['vmid'] print(f"🔄 VNC 티켓 새로고침 요청: {node}/{vmid}") # Proxmox 클라이언트 생성 client = ProxmoxClient( host=config['proxmox']['host'], username=config['proxmox']['username'], password=config['proxmox']['password'] ) if not client.login(): return jsonify({'error': 'Proxmox 로그인 실패'}), 500 # 새 VNC 티켓 생성 vnc_data = client.get_vnc_ticket(node, vmid) if not vnc_data: return jsonify({'error': '새 VNC 티켓 생성 실패'}), 500 # 세션 업데이트 vnc_sessions[session_id].update({ 'websocket_url': vnc_data['websocket_url'], 'password': vnc_data.get('password', ''), 'updated_at': datetime.now() }) print(f"✅ VNC 티켓 새로고침 완료: {session_id}") return jsonify({ 'success': True, 'websocket_url': vnc_data['websocket_url'], 'password': vnc_data.get('password', ''), 'message': '새 VNC 티켓이 생성되었습니다.' }) except Exception as e: print(f"❌ VNC 티켓 새로고침 실패: {e}") return jsonify({'error': str(e)}), 500 @app.route('/vnc/') def vnc_console(session_id): """VNC 콘솔 페이지""" try: # 세션 확인 if session_id not in vnc_sessions: return render_template('error.html', error='유효하지 않은 VNC 세션입니다.'), 404 session_data = vnc_sessions[session_id] # 직접 WebSocket VNC 연결 (noVNC) - 간단한 버전으로 테스트 return render_template('vnc_simple.html', vm_name=session_data['vm_name'], vmid=session_data['vmid'], node=session_data['node'], websocket_url=session_data['websocket_url'], password=session_data.get('password', ''), vm_status=session_data.get('vm_status', 'unknown')) except Exception as e: print(f"❌ VNC 콘솔 오류: {e}") return render_template('error.html', error=str(e)), 500 @app.route('/api/vm/start', methods=['POST']) def api_vm_start(): """VM 시작 API""" try: data = request.get_json() node = data.get('node') vmid = int(data.get('vmid')) # Proxmox 클라이언트 생성 및 로그인 client = ProxmoxClient(PROXMOX_HOST, PROXMOX_USERNAME, PROXMOX_PASSWORD) if not client.login(): return jsonify({'error': 'Proxmox 서버 로그인 실패'}), 500 # VM 시작 success = client.start_vm(node, vmid) if success: return jsonify({'message': f'VM {vmid} 시작 명령을 전송했습니다.', 'success': True}) else: return jsonify({'error': 'VM 시작 실패'}), 500 except Exception as e: print(f"❌ VM 시작 오류: {e}") return jsonify({'error': str(e)}), 500 @app.route('/api/vm/stop', methods=['POST']) def api_vm_stop(): """VM 정지 API""" try: data = request.get_json() node = data.get('node') vmid = int(data.get('vmid')) # Proxmox 클라이언트 생성 및 로그인 client = ProxmoxClient(PROXMOX_HOST, PROXMOX_USERNAME, PROXMOX_PASSWORD) if not client.login(): return jsonify({'error': 'Proxmox 서버 로그인 실패'}), 500 # VM 정지 success = client.stop_vm(node, vmid) if success: return jsonify({'message': f'VM {vmid} 정지 명령을 전송했습니다.', 'success': True}) else: return jsonify({'error': 'VM 정지 실패'}), 500 except Exception as e: print(f"❌ VM 정지 오류: {e}") return jsonify({'error': str(e)}), 500 # 노드 삭제 API @app.route('/api/nodes//delete', methods=['DELETE']) def api_delete_node(node_id): """노드 삭제 API""" try: # Headscale CLI를 통해 노드 삭제 result = subprocess.run( ['docker', 'exec', 'headscale', 'headscale', 'nodes', 'delete', '-i', str(node_id), '--force'], capture_output=True, text=True, check=True ) # 삭제 성공 return jsonify({ 'success': True, 'message': f'노드 {node_id}가 성공적으로 삭제되었습니다.', 'output': result.stdout }) except subprocess.CalledProcessError as e: # Headscale CLI 오류 error_msg = e.stderr if e.stderr else e.stdout return jsonify({ 'success': False, 'error': f'노드 삭제 실패: {error_msg}' }), 400 except Exception as e: # 기타 오류 print(f"❌ 노드 삭제 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 # 사용자 관리 라우트 @app.route('/users') def users_list(): """사용자 관리 페이지""" return render_template('users/list.html') # 사용자 관리 API @app.route('/api/users', methods=['GET']) def api_get_users(): """Headscale 사용자 목록 조회""" try: # Headscale CLI를 통해 사용자 목록 조회 result = subprocess.run( ['docker', 'exec', 'headscale', 'headscale', 'users', 'list', '-o', 'json'], capture_output=True, text=True, check=True ) users_data = json.loads(result.stdout) # FARMQ 약국 정보와 매칭 (명시적으로 매핑된 것만) farmq_session = get_farmq_session() try: pharmacies = farmq_session.query(PharmacyInfo).all() # 명시적으로 headscale_user_name이 설정되고 해당 사용자가 실제로 존재하는 경우만 매핑 pharmacy_map = {} for p in pharmacies: if p.headscale_user_name and p.headscale_user_name.strip(): pharmacy_map[p.headscale_user_name] = p # 사용자별 노드 수 조회 for user in users_data: user_name = user.get('name', '') # 약국 정보 매칭 - 명시적으로 연결된 것만 pharmacy = pharmacy_map.get(user_name) user['pharmacy'] = { 'id': pharmacy.id, 'name': pharmacy.pharmacy_name, 'manager': pharmacy.manager_name } if pharmacy else None # 해당 사용자의 노드 수 조회 node_result = subprocess.run( ['docker', 'exec', 'headscale', 'headscale', 'nodes', 'list', '-o', 'json'], capture_output=True, text=True, check=True ) nodes_data = json.loads(node_result.stdout) user['node_count'] = len([n for n in nodes_data if n.get('user', {}).get('name') == user_name]) return jsonify({'success': True, 'users': users_data}) finally: farmq_session.close() except subprocess.CalledProcessError as e: error_msg = e.stderr if e.stderr else e.stdout return jsonify({ 'success': False, 'error': f'사용자 목록 조회 실패: {error_msg}' }), 400 except Exception as e: print(f"❌ 사용자 목록 조회 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 @app.route('/api/users', methods=['POST']) def api_create_user(): """새 Headscale 사용자 생성""" try: data = request.get_json() user_name = data.get('name', '').strip() display_name = data.get('display_name', '').strip() email = data.get('email', '').strip() if not user_name: return jsonify({ 'success': False, 'error': '사용자 이름은 필수입니다.' }), 400 # Headscale CLI를 통해 사용자 생성 cmd = ['docker', 'exec', 'headscale', 'headscale', 'users', 'create', user_name] if display_name: cmd.extend(['-d', display_name]) if email: cmd.extend(['-e', email]) result = subprocess.run( cmd, capture_output=True, text=True, check=True ) return jsonify({ 'success': True, 'message': f'사용자 "{user_name}"가 성공적으로 생성되었습니다.', 'output': result.stdout }) except subprocess.CalledProcessError as e: error_msg = e.stderr if e.stderr else e.stdout return jsonify({ 'success': False, 'error': f'사용자 생성 실패: {error_msg}' }), 400 except Exception as e: print(f"❌ 사용자 생성 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 @app.route('/api/users//delete', methods=['DELETE']) def api_delete_user(user_name): """Headscale 사용자 삭제""" try: # Headscale CLI를 통해 사용자 삭제 result = subprocess.run( ['docker', 'exec', 'headscale', 'headscale', 'users', 'destroy', '--name', user_name, '--force'], capture_output=True, text=True, check=True ) return jsonify({ 'success': True, 'message': f'사용자 "{user_name}"가 성공적으로 삭제되었습니다.', 'output': result.stdout }) except subprocess.CalledProcessError as e: error_msg = e.stderr if e.stderr else e.stdout return jsonify({ 'success': False, 'error': f'사용자 삭제 실패: {error_msg}' }), 400 except Exception as e: print(f"❌ 사용자 삭제 오류: {e}") return jsonify({ 'success': False, 'error': f'서버 오류: {str(e)}' }), 500 # =================== 구독 서비스 관리 API =================== @app.route('/api/subscriptions/stats', methods=['GET']) def api_subscription_stats(): """대시보드용 구독 현황 통계""" try: import sqlite3 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 서비스별 구독 현황 cursor.execute(''' SELECT sp.product_code, sp.product_name, COUNT(ps.id) as subscription_count, SUM(ps.monthly_fee) as total_monthly_revenue FROM service_products sp LEFT JOIN pharmacy_subscriptions ps ON sp.id = ps.product_id AND ps.subscription_status = 'ACTIVE' GROUP BY sp.id, sp.product_code, sp.product_name ORDER BY sp.product_code ''') services = [] total_revenue = 0 total_subscriptions = 0 for row in cursor.fetchall(): product_code, product_name, count, revenue = row revenue = revenue or 0 services.append({ 'code': product_code, 'name': product_name, 'count': count, 'revenue': revenue }) total_revenue += revenue total_subscriptions += count # 전체 약국 수 cursor.execute("SELECT COUNT(*) FROM pharmacies") total_pharmacies = cursor.fetchone()[0] # 구독 중인 약국 수 cursor.execute(''' SELECT COUNT(DISTINCT pharmacy_id) FROM pharmacy_subscriptions WHERE subscription_status = 'ACTIVE' ''') subscribed_pharmacies = cursor.fetchone()[0] conn.close() return jsonify({ 'success': True, 'data': { 'services': services, 'total_revenue': total_revenue, 'total_subscriptions': total_subscriptions, 'total_pharmacies': total_pharmacies, 'subscribed_pharmacies': subscribed_pharmacies, 'subscription_rate': round(subscribed_pharmacies / total_pharmacies * 100, 1) if total_pharmacies > 0 else 0 } }) except Exception as e: print(f"❌ 구독 통계 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/pharmacies/subscriptions', methods=['GET']) def api_pharmacy_subscriptions(): """약국별 구독 현황 조회""" try: import sqlite3 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 약국별 구독 현황 조회 cursor.execute(''' SELECT p.id, p.pharmacy_name, p.manager_name, p.address, GROUP_CONCAT(sp.product_code) as subscribed_services, SUM(ps.monthly_fee) as total_monthly_fee FROM pharmacies p LEFT JOIN pharmacy_subscriptions ps ON p.id = ps.pharmacy_id AND ps.subscription_status = 'ACTIVE' LEFT JOIN service_products sp ON ps.product_id = sp.id GROUP BY p.id, p.pharmacy_name, p.manager_name, p.address ORDER BY total_monthly_fee DESC NULLS LAST, p.pharmacy_name ''') pharmacies = [] for row in cursor.fetchall(): pharmacy_id, name, manager, address, services, fee = row # 구독 서비스 리스트 변환 service_list = [] if services: for service_code in services.split(','): service_list.append(service_code) pharmacies.append({ 'id': pharmacy_id, 'name': name, 'manager': manager, 'address': address, 'subscribed_services': service_list, 'monthly_fee': fee or 0 }) conn.close() return jsonify({ 'success': True, 'data': pharmacies }) except Exception as e: print(f"❌ 약국 구독 현황 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/pharmacy//subscriptions', methods=['GET']) def api_pharmacy_subscription_detail(pharmacy_id): """특정 약국의 구독 상세 현황""" try: import sqlite3 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 약국 기본 정보 cursor.execute("SELECT pharmacy_name FROM pharmacies WHERE id = ?", (pharmacy_id,)) pharmacy_result = cursor.fetchone() if not pharmacy_result: return jsonify({'success': False, 'error': '약국을 찾을 수 없습니다.'}), 404 # 구독 중인 서비스 cursor.execute(''' SELECT ps.id, sp.product_code, sp.product_name, ps.monthly_fee, ps.start_date, ps.next_billing_date, ps.subscription_status, ps.notes FROM pharmacy_subscriptions ps JOIN service_products sp ON ps.product_id = sp.id WHERE ps.pharmacy_id = ? AND ps.subscription_status = 'ACTIVE' ORDER BY sp.product_name ''', (pharmacy_id,)) active_subscriptions = [] for row in cursor.fetchall(): sub_id, code, name, fee, start, next_bill, status, notes = row active_subscriptions.append({ 'id': sub_id, 'code': code, 'name': name, 'monthly_fee': fee, 'start_date': start, 'next_billing_date': next_bill, 'status': status, 'notes': notes }) # 구독 가능한 서비스 (미구독 서비스) cursor.execute(''' SELECT sp.id, sp.product_code, sp.product_name, sp.monthly_price, sp.description FROM service_products sp WHERE sp.is_active = 1 AND sp.id NOT IN ( SELECT ps.product_id FROM pharmacy_subscriptions ps WHERE ps.pharmacy_id = ? AND ps.subscription_status = 'ACTIVE' ) ORDER BY sp.product_name ''', (pharmacy_id,)) available_services = [] for row in cursor.fetchall(): prod_id, code, name, price, desc = row available_services.append({ 'id': prod_id, 'code': code, 'name': name, 'monthly_price': price, 'description': desc }) conn.close() return jsonify({ 'success': True, 'data': { 'pharmacy_name': pharmacy_result[0], 'active_subscriptions': active_subscriptions, 'available_services': available_services } }) except Exception as e: print(f"❌ 약국 구독 상세 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/subscriptions', methods=['POST']) def api_create_subscription(): """새 구독 생성""" try: import sqlite3 from datetime import date, timedelta data = request.get_json() pharmacy_id = data.get('pharmacy_id') product_id = data.get('product_id') monthly_fee = data.get('monthly_fee') notes = data.get('notes', '') if not pharmacy_id or not product_id: return jsonify({'success': False, 'error': '필수 파라미터가 누락되었습니다.'}), 400 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 중복 구독 확인 cursor.execute(''' SELECT id FROM pharmacy_subscriptions WHERE pharmacy_id = ? AND product_id = ? AND subscription_status = 'ACTIVE' ''', (pharmacy_id, product_id)) if cursor.fetchone(): conn.close() return jsonify({'success': False, 'error': '이미 구독 중인 서비스입니다.'}), 400 # 상품 정보 조회 cursor.execute("SELECT monthly_price FROM service_products WHERE id = ?", (product_id,)) product_result = cursor.fetchone() if not product_result: conn.close() return jsonify({'success': False, 'error': '존재하지 않는 서비스입니다.'}), 404 # 월 구독료 설정 (사용자 지정 또는 기본 가격) if monthly_fee is None: monthly_fee = product_result[0] # 구독 생성 start_date = date.today() next_billing_date = start_date + timedelta(days=30) cursor.execute(''' INSERT INTO pharmacy_subscriptions (pharmacy_id, product_id, subscription_status, start_date, next_billing_date, monthly_fee, notes) VALUES (?, ?, 'ACTIVE', ?, ?, ?, ?) ''', (pharmacy_id, product_id, start_date.isoformat(), next_billing_date.isoformat(), monthly_fee, notes)) conn.commit() subscription_id = cursor.lastrowid conn.close() return jsonify({ 'success': True, 'message': '구독이 성공적으로 생성되었습니다.', 'subscription_id': subscription_id }) except Exception as e: print(f"❌ 구독 생성 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/subscriptions/', methods=['DELETE']) def api_cancel_subscription(subscription_id): """구독 해지""" try: import sqlite3 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 구독 존재 확인 cursor.execute("SELECT id FROM pharmacy_subscriptions WHERE id = ?", (subscription_id,)) if not cursor.fetchone(): conn.close() return jsonify({'success': False, 'error': '존재하지 않는 구독입니다.'}), 404 # 구독 상태를 CANCELLED로 변경 cursor.execute(''' UPDATE pharmacy_subscriptions SET subscription_status = 'CANCELLED', updated_at = CURRENT_TIMESTAMP WHERE id = ? ''', (subscription_id,)) conn.commit() conn.close() return jsonify({ 'success': True, 'message': '구독이 성공적으로 해지되었습니다.' }) except Exception as e: print(f"❌ 구독 해지 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 # =================== 구독 서비스 관리 페이지 =================== @app.route('/subscriptions') def subscriptions_page(): """구독 서비스 관리 페이지""" return render_template('subscriptions/list.html') # =================== 매출 대시보드 =================== @app.route('/revenue') def revenue_dashboard(): """매출 대시보드 페이지""" return render_template('revenue/dashboard.html') @app.route('/api/analytics/revenue/monthly', methods=['GET']) def api_monthly_revenue(): """월별 매출 통계""" try: import sqlite3 from datetime import datetime, timedelta import calendar db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 최근 12개월 매출 데이터 생성 current_date = datetime.now() monthly_data = [] for i in range(12): # i개월 전 날짜 계산 target_date = current_date - timedelta(days=30*i) year = target_date.year month = target_date.month month_name = calendar.month_name[month] # 해당 월의 구독 수 및 매출 계산 cursor.execute(''' SELECT COUNT(ps.id) as subscription_count, SUM(ps.monthly_fee) as total_revenue FROM pharmacy_subscriptions ps WHERE ps.subscription_status = 'ACTIVE' AND ps.start_date <= ? ''', (f'{year}-{month:02d}-28',)) result = cursor.fetchone() subscription_count = result[0] or 0 total_revenue = result[1] or 0 monthly_data.insert(0, { 'month': f'{year}-{month:02d}', 'month_name': f'{month_name[:3]} {year}', 'subscription_count': subscription_count, 'revenue': total_revenue }) conn.close() return jsonify({ 'success': True, 'data': monthly_data }) except Exception as e: print(f"❌ 월별 매출 통계 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/analytics/revenue/by-service', methods=['GET']) def api_revenue_by_service(): """서비스별 매출 통계""" try: import sqlite3 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 서비스별 매출 통계 cursor.execute(''' SELECT sp.product_code, sp.product_name, COUNT(ps.id) as subscription_count, SUM(ps.monthly_fee) as total_revenue FROM service_products sp LEFT JOIN pharmacy_subscriptions ps ON sp.id = ps.product_id AND ps.subscription_status = 'ACTIVE' GROUP BY sp.id, sp.product_code, sp.product_name ORDER BY total_revenue DESC NULLS LAST ''') services = [] total_revenue = 0 for row in cursor.fetchall(): product_code, product_name, count, revenue = row revenue = revenue or 0 total_revenue += revenue services.append({ 'code': product_code, 'name': product_name, 'subscription_count': count, 'revenue': revenue }) # 비율 계산 for service in services: service['percentage'] = round((service['revenue'] / total_revenue * 100), 1) if total_revenue > 0 else 0 conn.close() return jsonify({ 'success': True, 'data': { 'services': services, 'total_revenue': total_revenue } }) except Exception as e: print(f"❌ 서비스별 매출 통계 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/analytics/pharmacy-ranking', methods=['GET']) def api_pharmacy_ranking(): """약국별 구독료 순위""" try: import sqlite3 db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 약국별 구독료 순위 (상위 10개) cursor.execute(''' SELECT p.pharmacy_name, p.manager_name, COUNT(ps.id) as service_count, SUM(ps.monthly_fee) as total_monthly_fee FROM pharmacies p JOIN pharmacy_subscriptions ps ON p.id = ps.pharmacy_id WHERE ps.subscription_status = 'ACTIVE' GROUP BY p.id, p.pharmacy_name, p.manager_name ORDER BY total_monthly_fee DESC LIMIT 10 ''') rankings = [] for i, row in enumerate(cursor.fetchall(), 1): pharmacy_name, manager_name, service_count, total_fee = row rankings.append({ 'rank': i, 'pharmacy_name': pharmacy_name, 'manager_name': manager_name, 'service_count': service_count, 'monthly_fee': total_fee }) conn.close() return jsonify({ 'success': True, 'data': rankings }) except Exception as e: print(f"❌ 약국별 순위 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 @app.route('/api/analytics/subscription-trends', methods=['GET']) def api_subscription_trends(): """구독 트렌드 통계 (신규/해지)""" try: import sqlite3 from datetime import datetime, timedelta db_path = '/srv/headscale-setup/farmq-admin/farmq.db' conn = sqlite3.connect(db_path) cursor = conn.cursor() # 최근 6개월 구독 트렌드 trends = [] current_date = datetime.now() for i in range(6): target_date = current_date - timedelta(days=30*i) year = target_date.year month = target_date.month # 해당 월 신규 구독 수 cursor.execute(''' SELECT COUNT(*) FROM pharmacy_subscriptions WHERE strftime('%Y-%m', start_date) = ? ''', (f'{year}-{month:02d}',)) new_subscriptions = cursor.fetchone()[0] # 해당 월 해지 구독 수 cursor.execute(''' SELECT COUNT(*) FROM pharmacy_subscriptions WHERE subscription_status = 'CANCELLED' AND strftime('%Y-%m', updated_at) = ? ''', (f'{year}-{month:02d}',)) cancelled_subscriptions = cursor.fetchone()[0] trends.insert(0, { 'month': f'{year}-{month:02d}', 'new_subscriptions': new_subscriptions, 'cancelled_subscriptions': cancelled_subscriptions, 'net_growth': new_subscriptions - cancelled_subscriptions }) conn.close() return jsonify({ 'success': True, 'data': trends }) except Exception as e: print(f"❌ 구독 트렌드 조회 오류: {e}") return jsonify({'success': False, 'error': str(e)}), 500 # 에러 핸들러 @app.errorhandler(404) def not_found_error(error): return render_template('error.html', error='페이지를 찾을 수 없습니다.', error_code=404), 404 @app.errorhandler(500) def internal_error(error): return render_template('error.html', error='내부 서버 오류가 발생했습니다.', error_code=500), 500 return app # 개발 서버 실행 if __name__ == '__main__': app = create_app() # 개발 환경에서만 실행 if app.config.get('DEBUG'): print("🚀 Starting FARMQ Admin System...") print(f"📊 Dashboard: http://localhost:5001") print(f"🏥 Pharmacy Management: http://localhost:5001/pharmacy") print(f"💻 Machine Management: http://localhost:5001/machines") print(f"🖥️ VM Management (VNC): http://localhost:5001/vms") print("─" * 60) app.run( host='0.0.0.0', port=5001, debug=True, use_reloader=True )