firstgarden/fg-auto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

성능 향상

Browse Source
This commit is contained in:
KWON
2025-06-27 10:47:23 +09:00
parent a5a1da6770
commit 82958bd843
1 changed files with 200 additions and 146 deletions
Download Patch File Download Diff File Expand all files Collapse all files

346
data/gnu_autoupload.py
View File

@ -1,180 +1,234 @@
import pymysql, ftputil, hashlib, os, sys
# -*- coding: utf-8 -*-
"""
gnu_autoupload.py
기능:
1. Selenium을 이용해 날씨 정보를 캡처 (weather_capture.py 호출)
2. FTP를 이용해 이미지 업로드
3. 그누보드 DB에 게시글 및 첨부파일 정보 자동 등록
"""
import os
import sys
import time
import subprocess
import tempfile
import hashlib
from datetime import datetime
from PIL import Image
import subprocess
import time
import os
import pymysql
import ftputil
from config import DB_CONFIG, FTP_CONFIG, MAIN
# 오늘 날짜 기반으로 파일명 생성
today_str = datetime.today().strftime('%Y%m%d')
base_dir = os.path.dirname(os.path.abspath(__file__)) # 현재 파이썬 파일의 경로
capture_script = os.path.join(base_dir, 'weather_capture.py')
weather_filename = f'weather_capture_{today_str}.png'
weather_file = os.path.join(base_dir, weather_filename)
# 파일 존재 확인 및 생성 루프
while not os.path.isfile(weather_file):
print(f"[{datetime.now().strftime('%H:%M:%S')}] {weather_file} 파일이 없습니다. {capture_script} 실행 중...")
subprocess.call(['python3', capture_script])
#time.sleep(1)
print(f"[{datetime.now().strftime('%H:%M:%S')}] {weather_file} 파일을 확인했습니다. 계속 진행합니다.")
# 파일 업로드 경로 설정
FTP_CONFIG['UPLOAD_DIR'] = f"/www/data/file/{MAIN['board']}/"
# 게시글 제목을 오늘 날짜를 반영하도록 수정
MAIN['subject'] = f"[{datetime.now().strftime('%Y-%m-%d')}] 날씨 정보"
MAIN['file1'] = MAIN['file2'] = weather_file
# ---------------------------
# 이미지 캡처 관련 함수
# ---------------------------
def capture_image(script_path, output_path, max_attempts=5):
"""
weather_capture.py를 실행해 이미지 파일을 생성
:param script_path: 캡처 스크립트 경로
:param output_path: 기대되는 이미지 파일 경로
:param max_attempts: 최대 재시도 횟수
:return: 성공 여부 (True/False)
"""
for attempt in range(max_attempts):
print(f"[{datetime.now().strftime('%H:%M:%S')}] 캡처 시도 {attempt + 1}/{max_attempts}")
try:
subprocess.run(['python3', script_path], check=True)
except subprocess.CalledProcessError as e:
print(f"[오류] weather_capture.py 실행 실패: {e}")
if os.path.isfile(output_path):
print(f"[성공] 이미지 캡처 완료: {output_path}")
return True
time.sleep(2)
print(f"[실패] {max_attempts}회 시도 후에도 이미지가 생성되지 않았습니다.")
return False
# ---------------------------
# 파일 업로드 및 처리 관련
# ---------------------------
def file_type(ext):
"""
파일 확장자를 기반으로 그누보드용 파일 유형 코드 반환
"""
return {
'gif': '1', 'jpeg': '2', 'jpg': '2', 'png': '3', 'swf': '4',
'psd': '5', 'bmp': '6', 'tif': '7', 'tiff': '7', 'jpc': '9',
'jp2': '10', 'jpx': '11', 'jb2': '12', 'swc': '13', 'iff': '14',
'wbmp': '15', 'xbm': '16'
}.get(ext.lower(), '0')
def file_type(x): # 그누보드의 bf_type 값을 반환하는 함수입니다. (디폴트 : 0)
return {'gif' : '1', 'jpeg' : '2', 'jpg' : '2', 'png' : '3', 'swf' : '4', 'psd' : '5',
'bmp' : '6', 'tif' : '7', 'tiff' : '7', 'jpc' : '9', 'jp2' : '10', 'jpx' : '11',
'jb2' : '12', 'swc' : '13', 'iff' : '14', 'wbmp' : '15', 'xbm' : '16'}.get(x.lower(), '0')
def file_upload(filename, bf_file): # FTP를 이용하여 파일을 업로드하는 함수입니다.
def get_filename(filename):
"""
원본 파일명을 기반으로 해시된 파일명을 생성
"""
ms = datetime.now().microsecond
encoded_name = filename.encode('utf-8')
return f'{ms}_{hashlib.sha1(encoded_name).hexdigest()}'
def file_upload(filename, bf_file):
"""
FTP 서버에 파일 업로드
:param filename: 로컬 경로
:param bf_file: 서버에 저장될 파일명
"""
try:
with ftputil.FTPHost(FTP_CONFIG['HOST'], FTP_CONFIG['USER'], FTP_CONFIG['PASS']) as fh:
fh.chdir(FTP_CONFIG['UPLOAD_DIR'])
fh.upload(filename, bf_file, callback = None)
print(f"[업로드 완료] {filename}")
fh.upload(filename, bf_file)
print(f"[업로드 완료] {filename} → {bf_file}")
return True
except Exception as e:
print(f"[FTP 오류] {type(e).__name__}: {e}")
return False
return
def get_filename(filename): # 파일명을 변환하는 함수입니다.
ms = datetime.now().microsecond
encoded_name = filename.encode('utf-8')
result = f'{ms}_{hashlib.sha1(encoded_name).hexdigest()}'
return result
def board_write(board, subject, content, mb_id, nickname, ca_name=None, file_list=None):
conn = pymysql.connect(host=DB_CONFIG['HOST'],
user=DB_CONFIG['USER'],
db=DB_CONFIG['DBNAME'],
password=DB_CONFIG['PASS'],
charset='utf8')
curs = conn.cursor()
# ---------------------------
# 게시글 작성 함수
# ---------------------------
def write_board(board, subject, content, mb_id, nickname, ca_name=None, file_list=None):
"""
그누보드 DB에 게시글과 파일을 한 트랜잭션으로 삽입
:param board: 게시판 ID
:param subject: 제목
:param content: 내용
:param mb_id: 회원 ID
:param nickname: 작성자 이름
:param ca_name: 카테고리명
:param file_list: 첨부파일 경로 리스트
"""
try:
# 데이터베이스 연결
conn = pymysql.connect(
host=DB_CONFIG['HOST'],
user=DB_CONFIG['USER'],
db=DB_CONFIG['DBNAME'],
password=DB_CONFIG['PASS'],
charset='utf8'
)
curs = conn.cursor()
now = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
sql = f"SELECT wr_num FROM g5_write_{board}"
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
conn.commit()
wr_num = str(int(curs.fetchone()[0]) - 1)
print(f"[결과] wr_num: {wr_num}")
# 게시글 번호(wr_num) 계산
curs.execute(f"SELECT wr_num FROM g5_write_{board}")
wr_num = str(int(curs.fetchone()[0]) - 1)
now = datetime.today().strftime('%Y-%m-%d %H:%M:%S')
# 게시글 삽입
curs.execute(f"""INSERT INTO g5_write_{board} SET wr_num = {wr_num},
wr_reply = '', wr_comment = 0, ca_name = %s, wr_option = 'html1', wr_subject = %s,
wr_content = %s, wr_link1 = '', wr_link2 = '', wr_link1_hit = 0, wr_link2_hit = 0,
wr_hit = 1, wr_good = 0, wr_nogood = 0, mb_id = %s, wr_password = '',
wr_name = %s, wr_email = '', wr_homepage = '', wr_datetime = %s, wr_last = %s,
wr_ip = '127.0.0.1'""",
(ca_name, subject, content, mb_id, nickname, now, now))
sql = f"""INSERT INTO g5_write_{board} SET wr_num = {wr_num},
wr_reply = '', wr_comment = 0, ca_name = '{ca_name}', wr_option = 'html1', wr_subject = '{subject}',
wr_content = '{content}', wr_link1 = '', wr_link2 = '',
wr_link1_hit = 0, wr_link2_hit = 0, wr_hit = 1, wr_good = 0, wr_nogood = 0,
mb_id = '{mb_id}', wr_password = '', wr_name = '{nickname}', wr_email = '', wr_homepage = '',
wr_datetime = '{now}', wr_last = '{now}', wr_ip = '111.111.111.111',
wr_comment_reply = '', wr_facebook_user = '', wr_twitter_user = '',
wr_1 = '', wr_2 = '', wr_3 = '', wr_4 = '', wr_5 = '', wr_6 = '', wr_7 = '', wr_8 = '', wr_9 = '', wr_10 = ''
"""
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
conn.commit()
print("[결과] 게시글 INSERT 완료")
# wr_id 조회
curs.execute(f"SELECT wr_id FROM g5_write_{board} ORDER BY wr_id DESC LIMIT 1")
wr_id = str(curs.fetchone()[0])
sql = f"SELECT wr_id FROM g5_write_{board}"
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
wr_id = str(curs.fetchall()[-1][0])
conn.commit()
print(f"[결과] wr_id: {wr_id}")
# wr_parent 업데이트
curs.execute(f"UPDATE g5_write_{board} SET wr_parent = {wr_id} WHERE wr_id = {wr_id}")
sql = f"UPDATE g5_write_{board} SET wr_parent = {wr_id} WHERE wr_id = {wr_id}"
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
conn.commit()
print("[결과] wr_parent 업데이트 완료")
# 새글 테이블 등록
curs.execute(f"""INSERT INTO g5_board_new (bo_table, wr_id, wr_parent, bn_datetime, mb_id)
VALUES (%s, %s, %s, %s, %s)""", (board, wr_id, wr_id, now, mb_id))
sql = f"""INSERT INTO g5_board_new (bo_table, wr_id, wr_parent, bn_datetime, mb_id)
VALUES ('{board}', '{wr_id}', '{wr_id}', '{now}', '{mb_id}')"""
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
conn.commit()
print("[결과] 새글 INSERT 완료")
# 게시글 수 증가
curs.execute(f"SELECT bo_count_write FROM g5_board WHERE bo_table = %s", (board,))
bo_count_write = int(curs.fetchone()[0])
curs.execute(f"UPDATE g5_board SET bo_count_write = %s WHERE bo_table = %s",
(bo_count_write + 1, board))
sql = f"SELECT bo_count_write FROM g5_board WHERE bo_table = '{board}'"
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
bo_count_write = str(int(curs.fetchone()[0]))
conn.commit()
print(f"[결과] 현재 게시글 수: {bo_count_write}")
# 파일 업로드 처리
file_count = 0
if file_list:
for idx, file in enumerate(file_list):
ext = os.path.splitext(file)[1].lstrip('.')
bf_file = f"{get_filename(file)}.{ext}"
sql = f"UPDATE g5_board SET bo_count_write = {bo_count_write} + 1 WHERE bo_table = '{board}'"
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
conn.commit()
print("[결과] 게시글 수 증가 완료")
if file_upload(file, bf_file):
img_type = file_type(ext)
width, height = (0, 0)
if img_type != '0':
with Image.open(file) as img:
width, height = img.size
if not file_list:
print("[종료] 첨부파일 없음. 연결 종료.")
conn.close()
sys.exit()
size = os.path.getsize(file)
curs.execute(f"""INSERT INTO g5_board_file SET bo_table = %s, wr_id = %s, bf_no = %s,
bf_source = %s, bf_file = %s, bf_content = '', bf_download = 0,
bf_filesize = %s, bf_width = %s, bf_height = %s, bf_type = %s, bf_datetime = %s""",
(board, wr_id, idx, os.path.basename(file), bf_file,
size, width, height, img_type, now))
file_count += 1
else:
print(f"[경고] 파일 업로드 실패: {file}")
file_count = len(file_list)
for cnt, file in enumerate(file_list):
ext = os.path.splitext(file)[1].lstrip('.')
bf_file = f'{get_filename(file)}.{ext}'
file_upload(file, bf_file)
type = file_type(ext)
# 첨부파일 수 업데이트
curs.execute(f"UPDATE g5_write_{board} SET wr_file = %s WHERE wr_id = %s", (file_count, wr_id))
if type != '0':
im = Image.open(file)
width, height = im.size
else:
width, height = 0, 0
size = os.path.getsize(file)
sql = f"""INSERT INTO g5_board_file SET bo_table = '{board}', wr_id = '{wr_id}',
bf_no = '{cnt}', bf_source = '{weather_filename}', bf_file = '{bf_file}',
bf_content = '', bf_download = 0, bf_filesize = '{size}',
bf_width = '{width}', bf_height = '{height}', bf_type = '{type}', bf_datetime = '{now}'"""
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
# 트랜잭션 커밋 (단 1회)
conn.commit()
print(f"[결과] 파일({file}) 업로드 정보 INSERT 완료")
print("[완료] 게시글 작성 및 파일 업로드 완료")
sql = f"UPDATE g5_write_{board} SET wr_file = '{file_count}' WHERE wr_id = '{wr_id}'"
print(f"[쿼리 실행] {sql}")
curs.execute(sql)
conn.commit()
print(f"[결과] 첨부파일 수 wr_file = {file_count} 업데이트 완료")
except Exception as e:
# 예외 발생 시 롤백
conn.rollback()
print(f"[DB 오류] {type(e).__name__}: {e}")
finally:
# 연결 종료
conn.close()
conn.close()
print("[종료] MySQL 연결 종료")
# ---------------------------
# 메인 실행 함수
# ---------------------------
def main():
board = MAIN['board']
ca_name = MAIN['ca_name']
subject = MAIN['subject']
content = MAIN['content']
mb_id = MAIN['mb_id']
nickname = MAIN['nickname']
file_list = [MAIN['file1'], MAIN['file2']]
# file_list = [MAIN['file1']]
board_write(board, subject, content, mb_id, nickname, ca_name, file_list)
# 파일 경로 및 이름 설정
today = datetime.today().strftime('%Y%m%d')
script_dir = os.path.dirname(os.path.abspath(__file__))
capture_script = os.path.join(script_dir, 'weather_capture.py')
weather_filename = f'weather_capture_{today}.png'
weather_file = os.path.join(script_dir, weather_filename)
# 이미지 생성 시도
if not capture_image(capture_script, weather_file):
return
# 설정값 동적 갱신
FTP_CONFIG['UPLOAD_DIR'] = f"/www/data/file/{MAIN['board']}/"
MAIN['subject'] = f"[{datetime.now().strftime('%Y-%m-%d')}] 날씨 정보"
MAIN['file1'] = MAIN['file2'] = weather_file
file_list = [MAIN['file1']]
# 게시글 작성
write_board(
board=MAIN['board'],
subject=MAIN['subject'],
content=MAIN['content'],
mb_id=MAIN['mb_id'],
nickname=MAIN['nickname'],
ca_name=MAIN['ca_name'],
file_list=file_list
)
# 캡처 파일 삭제
try:
if os.path.isfile(weather_file):
os.remove(weather_file)
print(f"[삭제 완료] {weather_file}")
except Exception as e:
print(f"[삭제 오류] {type(e).__name__}: {e}")
# ---------------------------
# 스크립트 실행 진입점
# ---------------------------
if __name__ == "__main__":
main()
# main() 완료 후 이미지 파일 삭제
if os.path.isfile(weather_file):
try:
os.remove(weather_file)
print(f"[삭제 완료] {weather_file} 파일을 삭제했습니다.")
except Exception as e:
print(f"[삭제 오류] {type(e).__name__}: {e}")
else:
print(f"[파일 없음] {weather_file} 파일을 찾을 수 없습니다.")