From 2ba2ffe890061b01d608203d8c43117949c7984e Mon Sep 17 00:00:00 2001
From: songtianlun <tianlun.song@foxmail.com>
Date: Wed, 4 Sep 2024 18:18:33 +0800
Subject: [PATCH] add migrate typecho

---
 migrate-blog/hexo_to_typecho/migrate.py       | 131 ++++++++++++++++++
 .../hexo_to_typecho/test_pytypecho.py         |  13 ++
 2 files changed, 144 insertions(+)
 create mode 100644 migrate-blog/hexo_to_typecho/migrate.py
 create mode 100644 migrate-blog/hexo_to_typecho/test_pytypecho.py

diff --git a/migrate-blog/hexo_to_typecho/migrate.py b/migrate-blog/hexo_to_typecho/migrate.py
new file mode 100644
index 0000000..9d11a0e
--- /dev/null
+++ b/migrate-blog/hexo_to_typecho/migrate.py
@@ -0,0 +1,131 @@
+import os
+import frontmatter
+from pytypecho import Typecho
+from datetime import datetime
+
+# 配置部分
+hexo_posts_dir = '/home/songtianlun/Sync/Develop/frytea/source/_posts'  # Hexo 文章目录
+wordpress_url = "https://blog.lnf1.skybyte.me/xmlrpc.php"
+wordpress_username = 'songtianlun'
+wordpress_password = 'sotilu,WP2024'
+
+typecho_url = "https://oskyla.com/action/xmlrpc"
+typecho_username = 'songtianlun'
+typecho_password = 'sotilu,typecho2019'
+
+tp = Typecho(typecho_url, typecho_username, typecho_password)
+
+# 上传统计信息
+total_posts = 0
+successful_posts = 0
+failed_posts = 0
+duplicate_posts = []
+
+
+def get_all_posts():
+    print("Fetching existing posts from WordPress...")
+    max_posts = 1000
+    all_posts = tp.get_posts(max_posts=max_posts)
+    print(f"Total existing posts fetched: {len(all_posts)}")
+    return {post.title: post.postid for post in all_posts}
+
+
+def scan_directory_for_posts(directory, category_prefix=""):
+    posts = []
+    post_count = 1
+    for root, dirs, files in os.walk(directory):
+        category = category_prefix + os.path.basename(root)  # 使用文件夹作为类别
+        for file in files:
+            if file.endswith('.md'):
+                file_path = os.path.join(root, file)
+                post_title, post_data = process_md_file(file_path)
+                if post_title and post_data:
+                    posts.append((post_title, post_data, category, post_count))
+                    post_count += 1
+    return posts
+
+
+def process_md_file(file_path):
+    with open(file_path, 'r', encoding='utf-8') as f:
+        post_data = frontmatter.load(f)
+
+    # 构造 WordPress 文章
+    post_title = post_data['title'] if 'title' in post_data else os.path.basename(file_path)
+    return post_title, post_data
+
+
+def create_typecho_post(post_title, post_data, category):
+    post = Typecho.Post(
+        title=post_title,
+        categories=[category],
+
+    )
+    post.title = post_title
+    post.content = post_data.content
+    post.terms_names = {
+        'category': [category],
+        'post_tag': post_data['tags'] if 'tags' in post_data else [],
+    }
+
+    if 'date' in post_data:
+        post.date = datetime.strptime(str(post_data['date']), '%Y-%m-%d %H:%M:%S')
+
+    post.post_status = 'publish'  # 或 'draft' 保存为草稿
+
+    return post
+
+
+def upload_post_with_retries(post, post_number, retries=3):
+    global successful_posts, failed_posts
+    attempt = 0
+    while attempt < retries:
+        try:
+            post_id = wp.call(NewPost(post))
+            print(f"Post {post_number}: '{post.title}' - Upload successful. WordPress ID: {post_id}")
+            successful_posts += 1
+            return True
+        except Exception as e:
+            attempt += 1
+            print(f"Post {post_number}: '{post.title}' - Attempt {attempt} failed with error: {e}")
+            if attempt == retries:
+                print(f"Post {post_number}: '{post.title}' - Failed to upload after {retries} attempts.")
+                failed_posts += 1
+                return False
+
+
+def upload_posts_to_wordpress(posts_to_upload):
+    global total_posts
+    total_posts = len(posts_to_upload)
+    for post, post_number in posts_to_upload:
+        try:
+            upload_post_with_retries(post, post_number)
+        except Exception:
+            print(f"Post {post_number}: '{post.title}' - Failed permanently.")
+
+
+if __name__ == "__main__":
+    existing_wp_posts = get_all_posts()
+    posts = scan_directory_for_posts(hexo_posts_dir)
+    posts_to_upload = []
+
+    for post_title, post_data, category, post_number in posts:
+        if post_title in existing_wp_posts:
+            print(f"Post '{post_title}' skipped: already exists.")
+            duplicate_posts.append((post_title, post_number))
+        else:
+            post = create_wordpress_post(post_title, post_data, category)
+            posts_to_upload.append((post, len(posts_to_upload) + 1))
+
+    upload_posts_to_wordpress(posts_to_upload)
+
+    print("Migration completed!")
+    print(f"Total posts processed: {total_posts}")
+    print(f"Successfully uploaded: {successful_posts}")
+    print(f"Failed uploads: {failed_posts}")
+    if duplicate_posts:
+        print("\nDuplicate posts found:")
+        for title, post_number in duplicate_posts:
+            print(f"Title: {title} - File Number: {post_number}")
+    else:
+        print("\nNo duplicate posts found.")
+
diff --git a/migrate-blog/hexo_to_typecho/test_pytypecho.py b/migrate-blog/hexo_to_typecho/test_pytypecho.py
new file mode 100644
index 0000000..e6f8e5f
--- /dev/null
+++ b/migrate-blog/hexo_to_typecho/test_pytypecho.py
@@ -0,0 +1,13 @@
+from pytypecho import Typecho
+
+typecho_url = "https://oskyla.com/action/xmlrpc"
+typecho_username = 'songtianlun'
+typecho_password = 'sotilu,typecho2019'
+
+te = Typecho(typecho_url, typecho_username, typecho_password)
+posts = te.get_posts(10)
+#print(posts)
+for post in posts:
+    print(post.get('title', "Untitled"), " - ", post.get('postid'))
+
+print(len(posts))
\ No newline at end of file