export_wp_posts.py
· 3.0 KiB · Python
Исходник
import os
import pymysql.cursors
import html2text as ht
host = 'localhost'
port = 3306
username = 'root'
password = 'root'
database = 'wordpress'
# Connect to the database
connection = pymysql.connect(host=host,
port=port,
user=username,
password=password,
db=database)
work_path = os.getcwd() + os.path.sep + "posts"
if not os.path.exists(work_path):
os.makedirs(work_path)
print("文件将会导出到脚本同目录下的'posts'文件夹中")
hexo_md_head = """---
title: %s
category: auto_export
comments: false
comment: false
date: %s
updated:
tags:
permalink:
thumbnail:
toc:
top:
mathJax:
---\n\n
%s
"""
# 查询所有文章
sql_all = """SELECT `ID`,`post_date_gmt`,`post_title`,`post_status`,`post_modified_gmt`,`post_type`,`post_content`
FROM `wp_posts`
WHERE `post_parent`=0
AND `post_type`='post'
ORDER BY `post_date_gmt` DESC
"""
# 根据ID查询文章最新版本
sql_post_newest = """SELECT `ID`,`post_date_gmt`,`post_title`,`post_status`,`post_modified_gmt`,`post_type`,`post_content`
FROM `wp_posts`
WHERE `post_parent`=%s
AND `post_status`='inherit'
ORDER BY `post_date_gmt`
DESC """
try:
with connection.cursor() as cursor:
cursor.execute(sql_all)
results = cursor.fetchall()
print("共查询到 %d 篇文章，即将开始导出..." % (results.__len__()))
htmlWorker = ht.HTML2Text()
htmlWorker.bypass_tables = False
for i, row in enumerate(results):
ID = row[0]
date = row[1]
title = row[2]
content = row[6]
print("正在导出第 %d 篇文章: ID=%d, date=%s, title=《%s》" % (i+1, ID, date, title))
cursor.execute(sql_post_newest % ID)
new_post = cursor.fetchone()
if new_post:
title = new_post[2]
content = new_post[6]
content = htmlWorker.handle(content)
try:
tmpPost = open(work_path + os.path.sep + str(date)[0:10] + "-"
+ str(title).replace(" ", "-")
.replace("/", "").replace("|","").replace("\\","").replace("\"","'")
.replace(":","").replace("*","").replace("?","").replace("#","")
.replace("<","(").replace(">",")").replace("【","[").replace("】","]")
+ ".md",
mode='w', encoding="utf-8")
tmpPost.write(hexo_md_head % (title, date, content))
tmpPost.flush()
tmpPost.close()
except:
err_post_info = "文章导出失败: ID=%d, title=《%s》 " % (ID, title)
print(err_post_info + "请查看日志")
export_log = open("." + os.path.sep + "error.log", mode='a', encoding='utf-8')
export_log.write(err_post_info + "\n\n")
export_log.close()
cursor.close()
finally:
connection.close()
| 1 | import os |
| 2 | import pymysql.cursors |
| 3 | import html2text as ht |
| 4 | |
| 5 | host = 'localhost' |
| 6 | port = 3306 |
| 7 | username = 'root' |
| 8 | password = 'root' |
| 9 | database = 'wordpress' |
| 10 | |
| 11 | # Connect to the database |
| 12 | connection = pymysql.connect(host=host, |
| 13 | port=port, |
| 14 | user=username, |
| 15 | password=password, |
| 16 | db=database) |
| 17 | |
| 18 | work_path = os.getcwd() + os.path.sep + "posts" |
| 19 | if not os.path.exists(work_path): |
| 20 | os.makedirs(work_path) |
| 21 | print("文件将会导出到脚本同目录下的'posts'文件夹中") |
| 22 | |
| 23 | hexo_md_head = """--- |
| 24 | title: %s |
| 25 | category: auto_export |
| 26 | comments: false |
| 27 | comment: false |
| 28 | date: %s |
| 29 | updated: |
| 30 | tags: |
| 31 | permalink: |
| 32 | thumbnail: |
| 33 | toc: |
| 34 | top: |
| 35 | mathJax: |
| 36 | ---\n\n |
| 37 | %s |
| 38 | """ |
| 39 | |
| 40 | # 查询所有文章 |
| 41 | sql_all = """SELECT `ID`,`post_date_gmt`,`post_title`,`post_status`,`post_modified_gmt`,`post_type`,`post_content` |
| 42 | FROM `wp_posts` |
| 43 | WHERE `post_parent`=0 |
| 44 | AND `post_type`='post' |
| 45 | ORDER BY `post_date_gmt` DESC |
| 46 | """ |
| 47 | |
| 48 | # 根据ID查询文章最新版本 |
| 49 | sql_post_newest = """SELECT `ID`,`post_date_gmt`,`post_title`,`post_status`,`post_modified_gmt`,`post_type`,`post_content` |
| 50 | FROM `wp_posts` |
| 51 | WHERE `post_parent`=%s |
| 52 | AND `post_status`='inherit' |
| 53 | ORDER BY `post_date_gmt` |
| 54 | DESC """ |
| 55 | |
| 56 | |
| 57 | try: |
| 58 | with connection.cursor() as cursor: |
| 59 | |
| 60 | cursor.execute(sql_all) |
| 61 | |
| 62 | results = cursor.fetchall() |
| 63 | print("共查询到 %d 篇文章，即将开始导出..." % (results.__len__())) |
| 64 | |
| 65 | htmlWorker = ht.HTML2Text() |
| 66 | htmlWorker.bypass_tables = False |
| 67 | |
| 68 | for i, row in enumerate(results): |
| 69 | ID = row[0] |
| 70 | date = row[1] |
| 71 | title = row[2] |
| 72 | content = row[6] |
| 73 | print("正在导出第 %d 篇文章: ID=%d, date=%s, title=《%s》" % (i+1, ID, date, title)) |
| 74 | |
| 75 | cursor.execute(sql_post_newest % ID) |
| 76 | new_post = cursor.fetchone() |
| 77 | if new_post: |
| 78 | title = new_post[2] |
| 79 | content = new_post[6] |
| 80 | |
| 81 | content = htmlWorker.handle(content) |
| 82 | |
| 83 | try: |
| 84 | tmpPost = open(work_path + os.path.sep + str(date)[0:10] + "-" |
| 85 | + str(title).replace(" ", "-") |
| 86 | .replace("/", "").replace("|","").replace("\\","").replace("\"","'") |
| 87 | .replace(":","").replace("*","").replace("?","").replace("#","") |
| 88 | .replace("<","(").replace(">",")").replace("【","[").replace("】","]") |
| 89 | + ".md", |
| 90 | mode='w', encoding="utf-8") |
| 91 | tmpPost.write(hexo_md_head % (title, date, content)) |
| 92 | tmpPost.flush() |
| 93 | tmpPost.close() |
| 94 | except: |
| 95 | err_post_info = "文章导出失败: ID=%d, title=《%s》 " % (ID, title) |
| 96 | print(err_post_info + "请查看日志") |
| 97 | export_log = open("." + os.path.sep + "error.log", mode='a', encoding='utf-8') |
| 98 | export_log.write(err_post_info + "\n\n") |
| 99 | export_log.close() |
| 100 | |
| 101 | cursor.close() |
| 102 | |
| 103 | finally: |
| 104 | connection.close() |