icurfer/wp-post-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

데모 코드 작성 완료

Browse Source
This commit is contained in:
icurfer 2024-10-04 00:02:07 +09:00
parent 019a1419d1
commit 7f03ff861b
12 changed files with 139 additions and 174 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
README.md
View File

@ -23,7 +23,11 @@ make.com을 이용해서 만든 AutoMation Flow를 Python을 이용하여 변환
* OpenAI이용 코드 작성(완료)-비용 절감을 위하여 제목, 이미지 생성 제외.
* HTML문서 변환 코드 작성(완료).
* 워드프레스 등록 플로우 코드 작성(기능 테스트 완료 - 연결 필요).
* 코드 리팩토링.
### 코드 리팩토링.
* 환경변수 호출 모듈분리(완료)
* 데이터베이스 조회 모듈분리(완료)
* OpenAI 모듈
## 코드 이슈
### 네이버 블로그 크롤링

33
main.py
View File

@ -1,12 +1,37 @@
from package import GetConfig
from package import MariaDB
import package as pkg
from package import GetConfig, MariaDB, ChangeTextToPost, WordPress
import markdown
# 2024-10-03 환경 변수 호출
config = GetConfig.GetConfig()
config = GetConfig()
dict_data = config.get_config_as_dict()
# 2024-10-03 db에서 url정보 호출
db = MariaDB.MariaDB(dict_data)
db = MariaDB(dict_data)
url = db.fetch_data_from_mariadb()['url']
# 2024-10-03 url을 이용해서 text추출
origin_content = pkg.getContents(url)
# 2024-10-03 openAI를 이용하여 게시글 스타일 변경
openai_key = dict_data['openai_api_key']
wp_reference_style = dict_data['wp_post_style']
open_ai = ChangeTextToPost(openai_key)
post_article = open_ai.generate_blog_post(origin_content, wp_reference_style)
# 2024-10-03 Markdown을 HTML로 변환
html = markdown.markdown(post_article)
# 2024-10-03 워드프레스 포스팅 임시등록
wp = WordPress(dict_data)
rs = wp.create_post(2,html)
if __name__ == "__main__":
# print(post_article)
print("추가 확인을 위한 출력")
if rs.ok:
print(f"성공 code:{rs.status_code}")
else:
print(f"실패 code:{rs.status_code} reason:{rs.reason} msg:{rs.text}")

2
package/GetConfig.py
View File

@ -9,7 +9,7 @@ class GetConfig:
self.db_user = os.getenv('DB_USER')
self.db_pw = os.getenv('DB_PASSWORD')
self.db_database = os.getenv('DB_NAME')
self.openai_api_key = os.getenv('OPEN_API_KEY')
self.openai_api_key = os.getenv('OPENAI_API_KEY')
self.wp_url = os.getenv('WP_URL')
self.wp_user = os.getenv('WP_USER')
self.wp_api_key = os.getenv('WP_API_KEY')

1
package/MariaDB.py
View File

@ -1,5 +1,4 @@
import mysql.connector
import os
class MariaDB:

37
package/OpenAI.py Normal file
View File

@ -0,0 +1,37 @@
from openai import OpenAI
class ChangeTextToPost:
def __init__(self, key):
self.client = OpenAI(
api_key=key,
)
def generate_blog_post(self, origin_content, wp_post_style):
# ChatCompletion API를 사용하여 텍스트 변환
prompt = (
"""
너는 대한민국에 거주하는 블로그 전문가이다.
네가 작성한 블로그 글은 지난 3년간 높은 주목성, 관여도, 전환율을 만들었다.
이 전문성을 이용해서 제공받는 기사를 블로그 형태로 변형하여 작성해야만 한다.
---
글을 작성하고 제목을 만들어서 맨 마지막 줄에 추가해줘.
"""
f"\n블로그 스타일은 아래 문서를 모방해줘. markdown형태로 작성되어야한다. 적절한 사례들이 들어가도 좋겠어.\n---\n{wp_post_style}\n"
f"제공된 기사 내용:\n{origin_content}"
)
response = self.client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "너는 대한민국에 거주하는 블로그 전문가이다."},
{"role": "user", "content": prompt}
],
max_tokens=2000,
temperature=0.7, # 창의성을 위한 적절한 값 조정
)
# 응답에서 텍스트 추출
blog_post = response.choices[0].message.content
return blog_post

61
package/Utility.py Normal file
View File

@ -0,0 +1,61 @@
import requests, json
from urllib.parse import urljoin
from bs4 import BeautifulSoup
from datetime import datetime
def getContents(url):
# HTTP GET 요청으로 페이지 가져오기
response = requests.get(url)
# 응답 상태 확인
if response.status_code == 200:
# HTML 파싱
soup = BeautifulSoup(response.text, 'html.parser')
# HTML 태그를 제거 후 페이지의 모든 텍스트 가져오기 (전체 내용)
page_content = soup.get_text()
# 빈 줄을 제거하고 텍스트만 출력 (줄바꿈 문자를 기준으로 필터링)
lines = [line.strip() for line in page_content.splitlines() if line.strip()]
# 결과 출력
contents = "\n".join(lines)
return contents
else:
print(f"Failed to fetch the URL. Status code: {response.status_code}")
class WordPress():
def __init__(self, dict):
self.wp_url = dict['wp_url']
self.wp_user = dict['wp_user']
self.wp_user = dict['wp_user']
self.wp_api_key = dict['wp_api_key']
def create_post(self, category_id, content, media_id = None, status = "draft", title="파이썬 자동 포스팅"):
payload = {
"status": status, # publish / draft
"title": title,
"content": content,
"date": datetime.now().isoformat(), # YYYY-MM-DDTHH:MM:SS
"categories": category_id
}
if media_id is not None:
payload['featured_media'] = media_id
return requests.post(urljoin(self.wp_url, "wp-json/wp/v2/posts"),
data=json.dumps(payload),
headers={'Content-type': "application/json"},
auth=(self.wp_user, self.wp_api_key))
# if result.ok:
# print(f"성공 code:{result.status_code}")
# else:
# print(f"실패 code:{result.status_code} reason:{result.reason} msg:{result.text}")
if __name__ == "__main__":
# url = 'example_url'
# tmp = getContents(url)
# print(tmp)
pass

7
package/__init__.py
View File

@ -1,2 +1,5 @@
from . import GetConfig
from . import MariaDB
from .Utility import getContents
from .Utility import WordPress
from .GetConfig import GetConfig
from .MariaDB import MariaDB
from .OpenAI import ChangeTextToPost

2
package/convert_md.py
View File

@ -2,7 +2,7 @@ import os
from dotenv import load_dotenv
import markdown
import package.translate_article as ta
import package.open_ai as oa
import package.OpenAI as oa
# .env 파일에서 API 키 로드
load_dotenv(r'./.env.dev')

38
package/get_url.py
View File

@ -1,38 +0,0 @@
import mysql.connector
import os
# MariaDB에 연결하는 함수
def fetch_data_from_mariadb():
try:
# 데이터베이스 연결
connection = mysql.connector.connect(
host=host,
user=user,
password=password,
database=database
)
# 커서 생성
cursor = connection.cursor(dictionary=True)
# 쿼리 실행
query = "SELECT * FROM healty_url_source ORDER BY idx DESC LIMIT 1;"
cursor.execute(query)
# 결과 가져오기
result = cursor.fetchone()
return result
except mysql.connector.Error as err:
print(f"Error: {err}")
finally:
if connection.is_connected():
cursor.close()
connection.close()
if __name__ == "__main__":
# 결과 확인
data = fetch_data_from_mariadb()
print(data['url'])

53
package/open_ai.py
View File

@ -1,53 +0,0 @@
import os
from openai import OpenAI
from dotenv import load_dotenv
import package.translate_article as ta
# .env 파일에서 API 키 로드
load_dotenv()
client = OpenAI(
api_key=os.getenv("OPENAI_API_KEY"),
)
def generate_blog_post(article_text, style_reference):
# ChatCompletion API를 사용하여 텍스트 변환
prompt = (
"""
너는 대한민국에 거주하는 블로그 전문가이다.
네가 작성한 블로그 글은 지난 3년간 높은 주목성, 관여도, 전환율을 만들었다.
이 전문성을 이용해서 제공받는 기사를 블로그 형태로 변형하여 작성해야만 한다.
---
글을 작성하고 제목을 만들어서 맨 마지막 줄에 추가해줘.
"""
f"\n블로그 스타일은 아래 문서를 모방해줘. 적절한 사례들이 들어가도 좋겠어.\n---\n{style_reference}\n"
f"제공된 기사 내용:\n{article_text}"
)
response = client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "너는 대한민국에 거주하는 블로그 전문가이다."},
{"role": "user", "content": prompt}
],
max_tokens=2000,
temperature=0.7, # 창의성을 위한 적절한 값 조정
)
# 응답에서 텍스트 추출
blog_post = response.choices[0].message.content
return blog_post
if __name__ == "__main__":
# 예시 기사 텍스트 (text 파싱 결과로 제공된 텍스트를 사용할 수 있습니다)
article_text = ta.getContents()
# 블로그 스타일 참고 텍스트
style_reference = os.getenv("reference_style")
# 블로그 포스트 생성
blog_post = generate_blog_post(article_text, style_reference)
# 결과 출력
print(">>>>\n" * 3)
print(blog_post)

30
package/translate_article.py
View File

@ -1,30 +0,0 @@
import requests
from bs4 import BeautifulSoup
import db_ctr.get_url as get_url
url = get_url.fetch_data_from_mariadb()['url']
def getContents():
# HTTP GET 요청으로 페이지 가져오기
response = requests.get(url)
# 응답 상태 확인
if response.status_code == 200:
# HTML 파싱
soup = BeautifulSoup(response.text, 'html.parser')
# HTML 태그를 제거 후 페이지의 모든 텍스트 가져오기 (전체 내용)
page_content = soup.get_text()
# 빈 줄을 제거하고 텍스트만 출력 (줄바꿈 문자를 기준으로 필터링)
lines = [line.strip() for line in page_content.splitlines() if line.strip()]
# 결과 출력
contents = "\n".join(lines)
return contents
else:
print(f"Failed to fetch the URL. Status code: {response.status_code}")
tmp = getContents()
print(tmp)

43
package/wp.py
View File

@ -1,43 +0,0 @@
import os
from dotenv import load_dotenv
import json
import requests
from urllib.parse import urljoin
from datetime import datetime
# .env 파일에서 API 키 로드
load_dotenv(r'./.env.dev')
wp_url = os.getenv('WP_URL')
wp_usr = os.getenv('WP_USERNAME')
wp_key = os.getenv('WP_API_KEY')
status = 'draft' # 즉시발행 publish, 임시저장 draft
# slug = 'input your slug'
title = '파이썬 자동포스팅'
content = 'html 호출 연동 필요.' # 연결만하면됨 테스트 완료.
category_id = [2] # 카테고리 아이디
# tag_ids = [21] #태그아이디
media_id=None #이미지 업로드를 제외 None
payload = {
"status": status,
"title": title,
"content": content,
"date": datetime.now().isoformat(), # YYYY-MM-DDTHH:MM:SS
"categories": category_id
}
if media_id is not None:
payload['featured_media'] = media_id
result = requests.post(urljoin(wp_url, "wp-json/wp/v2/posts"),
data=json.dumps(payload),
headers={'Content-type': "application/json"},
auth=(wp_usr, wp_key))
if result.ok:
print(f"성공 code:{result.status_code}")
else:
print(f"실패 code:{result.status_code} reason:{result.reason} msg:{result.text}")