Coverage for repositories / stock_code_repository.py: 93%

1# repositories/stock_code_repository.py 

2 

3import os 

4import sqlite3 

5import pandas as pd 

6from services.stock_sync_service import save_stock_code_list 

7 

8TABLE_NAME = "stocks" 

9 

10 

11def _write_minimal_db(db_path: str, logger=None): 

12 """빈/손상된 DB 대신 최소 유효 DB를 써서 앱이 시작되도록 합니다.""" 

13 os.makedirs(os.path.dirname(db_path), exist_ok=True) 

14 conn = sqlite3.connect(db_path) 

15 try: 

16 conn.execute(f"DROP TABLE IF EXISTS {TABLE_NAME}") 

17 conn.execute( 

18 f"CREATE TABLE {TABLE_NAME} (종목코드 TEXT, 종목명 TEXT, 시장구분 TEXT)" 

19 ) 

20 conn.execute( 

21 f"INSERT INTO {TABLE_NAME} VALUES (?, ?, ?)", 

22 ("000000", "(종목목록 없음)", ""), 

23 ) 

24 conn.commit() 

25 finally: 

26 conn.close() 

27 if logger: 

28 logger.warning(f"종목코드 DB가 비어 있어 최소 파일로 생성했습니다. 나중에 스크립트로 갱신하세요: {db_path}") 

29 

30 

31class StockCodeRepository: 

32 """ 

33 종목코드 ↔ 종목명 변환 기능을 제공하는 SQLite 기반 유틸리티 클래스. 

34 """ 

35 def __init__(self, db_path=None, logger=None): 

36 self.logger = logger 

37 if db_path is None: 

38 root = os.path.abspath(os.path.join(os.path.dirname(__file__), "..")) 

39 db_path = os.path.join(root, "data", "stock_code_list.db") 

40 self._db_path = db_path 

41 

42 # DB 파일이 없으면 생성 시도 

43 if not os.path.exists(db_path): 

44 if self.logger: 

45 self.logger.info(f"🔍 종목코드 매핑 DB 파일 없음. 생성 시작: {db_path}") 

46 try: 

47 save_stock_code_list(force_update=True) 

48 if self.logger: 48 ↛ 55line 48 didn't jump to line 55 because the condition on line 48 was always true

49 self.logger.info("✅ 종목코드 매핑 DB 파일 생성 완료.") 

50 except Exception as e: 

51 if self.logger: 

52 self.logger.error(f"❌ 종목코드 매핑 DB 파일 생성 실패: {e}") 

53 raise e 

54 

55 self._load_data() 

56 

57 def _load_data(self): 

58 try: 

59 conn = sqlite3.connect(self._db_path) 

60 try: 

61 self.df = pd.read_sql( 

62 f"SELECT * FROM {TABLE_NAME}", conn, dtype={"종목코드": str} 

63 ) 

64 finally: 

65 conn.close() 

66 

67 if self.df.empty or len(self.df.columns) == 0 or (len(self.df) == 1 and self.df.iloc[0]["종목코드"] == "000000"): 67 ↛ 68line 67 didn't jump to line 68 because the condition on line 67 was never true

68 raise ValueError("DB 테이블이 비어있거나 최소 DB 상태입니다.") 

69 

70 self.code_to_name = dict(zip(self.df["종목코드"], self.df["종목명"])) 

71 self.name_to_code = dict(zip(self.df["종목명"], self.df["종목코드"])) 

72 if self.logger: 

73 self.logger.info(f"🔄 종목코드 매핑 DB 로드 완료: {self._db_path}") 

74 except Exception as e: 

75 if self.logger: 75 ↛ 79line 75 didn't jump to line 79 because the condition on line 75 was always true

76 self.logger.warning(f"⚠️ 종목코드 DB 갱신/복구 시도 중 (사유: {e})") 

77 

78 # 손상된 기존 파일 삭제 시도 

79 try: 

80 if os.path.exists(self._db_path): 80 ↛ 88line 80 didn't jump to line 88 because the condition on line 80 was always true

81 os.remove(self._db_path) 

82 if self.logger: 82 ↛ 88line 82 didn't jump to line 88 because the condition on line 82 was always true

83 self.logger.info(f"🗑️ 손상된 DB 파일 삭제 완료: {self._db_path}") 

84 except Exception as remove_err: 

85 if self.logger: 85 ↛ 88line 85 didn't jump to line 88 because the condition on line 85 was always true

86 self.logger.error(f"❌ 손상된 DB 파일 삭제 실패 (파일 점유/권한 문제 등): {remove_err}") 

87 

88 try: 

89 save_stock_code_list(force_update=True) 

90 conn = sqlite3.connect(self._db_path) 

91 try: 

92 self.df = pd.read_sql( 

93 f"SELECT * FROM {TABLE_NAME}", conn, dtype={"종목코드": str} 

94 ) 

95 finally: 

96 conn.close() 

97 if self.df.empty or len(self.df.columns) == 0 or (len(self.df) == 1 and self.df.iloc[0]["종목코드"] == "000000"): 97 ↛ 98line 97 didn't jump to line 98 because the condition on line 97 was never true

98 raise ValueError("DB 테이블이 비어있거나 최소 DB 상태입니다.") 

99 self.code_to_name = dict(zip(self.df["종목코드"], self.df["종목명"])) 

100 self.name_to_code = dict(zip(self.df["종목명"], self.df["종목코드"])) 

101 if self.logger: 101 ↛ exitline 101 didn't return from function '_load_data' because the condition on line 101 was always true

102 self.logger.info(f"🔄 종목코드 매핑 DB 로드 완료: {self._db_path}") 

103 except Exception: 

104 if self.logger: 104 ↛ 106line 104 didn't jump to line 106 because the condition on line 104 was always true

105 self.logger.warning("갱신 실패. 최소 DB로 앱을 시작합니다.") 

106 _write_minimal_db(self._db_path, self.logger) 

107 conn = sqlite3.connect(self._db_path) 

108 try: 

109 self.df = pd.read_sql( 

110 f"SELECT * FROM {TABLE_NAME}", conn, dtype={"종목코드": str} 

111 ) 

112 finally: 

113 conn.close() 

114 self.code_to_name = dict(zip(self.df["종목코드"], self.df["종목명"])) 

115 self.name_to_code = dict(zip(self.df["종목명"], self.df["종목코드"])) 

116 

117 def get_name_by_code(self, code: str) -> str: 

118 name = self.code_to_name.get(code, "") 

119 if not name and self.logger: 

120 self.logger.warning(f"❗ 종목명 없음: {code}") 

121 return name 

122 

123 def get_code_by_name(self, name: str) -> str: 

124 code = self.name_to_code.get(name, "") 

125 if not code and self.logger: 

126 self.logger.warning(f"❗ 종목코드 없음: {name}") 

127 return code 

128 

129 def search_by_name(self, keyword: str, limit: int = 20) -> list: 

130 """종목명 부분 일치 검색. [{"code": "005930", "name": "삼성전자"}, ...] 형태로 반환.""" 

131 keyword_lower = keyword.lower() 

132 results = [] 

133 for name, code in self.name_to_code.items(): 

134 if keyword_lower in name.lower(): 

135 results.append({"code": code, "name": name}) 

136 if len(results) >= limit: 

137 break 

138 return results 

139 

140 def get_kosdaq_codes(self) -> list: 

141 """코스닥 시장 종목코드 리스트 반환.""" 

142 if "시장구분" not in self.df.columns: 

143 return [] 

144 return self.df[self.df["시장구분"] == "KOSDAQ"]["종목코드"].tolist() 

145 

146 def is_kosdaq(self, code: str) -> bool: 

147 """해당 종목코드가 코스닥 시장인지 확인.""" 

148 if "시장구분" not in self.df.columns: 

149 return False 

150 row = self.df[self.df["종목코드"] == code] 

151 return not row.empty and row.iloc[0]["시장구분"] == "KOSDAQ"