在我迁移的论坛中，曾经安装过 xengallery，因此我不得不更改以下内容，因为表 xfgallery 已不再存在。
def get_xf_sql(type, id)
case type
when :gallery
return "SELECT NULL WHERE 1=0;"
when :attachment
<<-SQL
SELECT a.attachment_id, a.data_id, d.filename, d.file_hash, d.user_id
FROM #{TABLE_PREFIX}attachment AS a
INNER JOIN #{TABLE_PREFIX}attachment_data d ON a.data_id = d.data_id
WHERE attachment_id = #{id}
AND content_type = 'post'
SQL
end
end
您可以尝试再次运行脚本，看看它是否完成。
您的系统是否在单个分区上？
mysql-server 安装步骤现已过时。需要安装 mariadb-server。
sudo apt-get install mariadb-server mariadb-client libmariadb-dev-compat libmariadb-dev
我能够按照其余步骤操作并导入 XF DB，因此有人应该更新指南。
我现在卡在以下步骤，需要帮助。
echo "gem 'mysql2'" >> Gemfile
bundle install --no-deployment
运行上述命令时出现以下错误。我检查了 Gemfile，它只包含这一行 - gem ‘mysql2’
此 Gemfile 未包含显式的全局源。
不使用显式全局源可能会导致在运行 bundler 之前根据您本地安装的 gem 生成不同的 lockfile。
请在此处 Gemfile 中定义全局源：source "https://rubygems.org"。
在本地安装的 gem 中找不到 gem 'mysql2'。
root@ip-172-566-459-13-app:/#
好的，我已经进入下一步。上面有人发帖说我们需要在容器的 /var/www/discourse 文件夹中，然后添加 gem。
现在是最后一步
RAILS_ENV=production bundle exec ruby script/import_scripts/xenforo.rb
我收到了这个错误。我可能做错了什么？
/var/www/discourse/vendor/bundle/ruby/3.3.0/gems/activerecord-7.2.2.1/lib/active_record/connection_adapters/postgresql_adapter.rb:63:in "rescue in new_client": We could not find your database: discourse. Available database configurations can be found in config/database.yml. (ActiveRecord::NoDatabaseError)To resolve this error:- Did you not create the database, or did you delete it? To create the database, run: bin/rails db:create- Has the database name changed? Verify that config/database.yml contains the correct database name.
已解决：我之前是以 root 用户运行的，必须切换到 ‘discourse’ 用户。导入已开始。
我租用了一个配置为 4CPU 和 16GB RAM 的相当不错的服务器。按照帖子迁移的速度，仅帖子迁移就需要 9 天。用户迁移花了 2.5 小时。可以肯定地说，目前对我来说这是不可行的，但我至少可以花几个月的时间熟悉它，直到我找到批量迁移的解决方案。
附注：
在迁移脚本中，我看到重复的电子邮件未被导入。确定重复的电子邮件有哪些不同的方法？我注意到 xyz@gmail.com 与 xyz+1@gmail.com 和 xy.z@gmail.com 被视为相同。
还有其他模式吗？
我曾在配置与我的个人电脑相似的 VPS 上进行迁移，但不知何故，速度总是比在我的电脑上慢得多。
现在，我总是本地进行迁移。您有多少帖子？
250 万条帖子。
将在 M1 Mac 上尝试本地迁移以进行比较。
差不多就是这样。唯一性检查是在给定电子邮件地址的转换为小写和规范化版本上执行的。
我们通过删除所有点并忽略用户名中 + 之后的所有内容来进行规范化。
单CPU速度是重要因素。
在我的机器上，每分钟800-1000个用户或帖子的速率是相当典型的。
请注意，当您进行最终导入时，它只会导入尚未导入的用户和帖子，因此不会花费很长时间。
关闭 Normalize emails 站点设置（直到最近它还是默认关闭的）。它可能需要在此函数中关闭：
您可以在您定制的 xenforo 脚本版本中通过 SiteSetting.normalize_emails=false 来设置。我不确定那些重复电子邮件的用户发生了什么；有两种明显的处理方法：给他们一个假的电子邮件地址或跳过导入他们。看起来是给了他们假的电子邮件地址？（而且他们很可能是假的用户的可能性相当大）。如果脚本跳过了他们，那么再次运行脚本将导入他们。
是的，在我的笔记本电脑上，它的处理速度快得多，每分钟可以处理 1000 个项目。这大约是服务器上的两倍快。不过，这仍然需要大约 3 天的时间。
我查看了被跳过的电子邮件，似乎它很好地忽略了那些账户。我将在最终导入前将它们合并。只有大约 20 几个这样的情况。
请注意，当您进行最终导入时，它只会导入尚未导入的用户和帖子，因此不会花费很长时间。
感谢您的提醒。我注意到了这一点，这似乎是在我进行最终导入时能够节省时间的关键。那么，我是否应该先进行一次备份并恢复到 D-3，然后再进行另一次备份并使用新的数据库备份文件恢复到第 0 天？是这样吗？
这些备份和恢复是在 Xenforo 网站上进行的，还是您有一个实时 Discourse 网站，您将把 Xenforo 数据导入到其中？
只要您不对脚本进行需要重新导入数据的更改，并且您笔记本电脑上的内容就是您想要在 Discourse 服务器上的内容，那么您就可以继续获取 Xenforo 数据库的新转储并导入它们（用于测试、查看需要多长时间等等），然后在切换日，冻结 Xenforo 网站，获取该数据库，再次运行脚本并上传到您的 Discourse 服务器。
如果您已经在 Discourse 网站上有想要保留的数据，那么情况会复杂得多，因为您需要冻结该网站，然后获取 Xenforo 数据，然后按上述方法进行。
这将是 Discourse 的全新安装，因此过程会很简单。
我有很多时间，因为我想多次测试迁移，彻底熟悉 Discourse，按照我想要的方式配置所有插件，甚至可能自己动手进行一些插件定制。
您所解释的内容完全消除了我的一块心病，因为我原以为我还得弄清楚批量导入的问题。
导入脚本是否会输出任何日志？我的测试导入已卡在 98.2% 几个小时了。
另外我发现，如果我重新启动迁移，它大约需要 30 秒才能跳过一批 1000 篇帖子。因此，现在的有效速度是每分钟 2000 个项目。与第一次导入的每分钟 1000 篇帖子相比，这并没有显著的改进，因为即使在切换当天的最后一次导入，也需要大约一天的时间。其中 23 小时将仅仅是跳过已导入的项目。
只有你看到的。
你应该停止它，然后重新启动它。
是的，它会跳过所有已导入的数据。而且它的速度比每分钟 2000 个帖子快得多。我怀疑你现在重新启动它时就会看到。
这就是我重启然后发了上面的帖子。它是每分钟 2000 帖。为了确定，我又试了一次。
已成功导入头像和附件。复制了这些文件夹。
/internal_data/attachments
/data/avatars
回答我的问题，头像和附件在导入后即定型。如果用户在导入其 ID 后更改了头像，则在第二次运行时将跳过该帖子或用户，因此不会导入/更新。
现在只需要弄清楚对话导入（也可以跳过，但最好有）和永久重定向。
@Fajfi - 感谢您为导入脚本做出的贡献。头像和附件导入完美运行。它仍在运行，尚未进行到点赞部分。
已修复对话导入。能够将超过五十万条消息从 XF2.3 导入到 discourse。已提交一个 PR，供有兴趣的人参考。
----编辑----
提交了另一个 PR，用于修复点赞导入。令人惊讶的是，到目前为止还没有人从 XF2.1+ 迁移到 discourse。点赞在 2019 年 XF2.1 发布时被重命名为反应。
我在导入附件时遇到了一些问题。
这表现为帖子中出现了 [attach]number[/attach]。
我认为这是由于哈希文件名有时与数据库记录不同步导致的。
我修改了这段代码：
def import_xf_attachment(data_id, file_hash, owner_id, original_filename)
matching_files = Dir.glob(File.join(ATTACHMENT_DIR, "**", "#{data_id}-*.data"))
filename = matching_files.first
upload = nil
if filename && File.exist?(filename)
path = Pathname.new(filename)
new_path = path.dirname.join(original_filename)
FileUtils.cp path, new_path
upload = @uploader.create_upload owner_id, new_path, original_filename
FileUtils.rm(new_path) if File.exist?(new_path)
else
STDERR.puts "Could not find file for data_id #{data_id}. Skipping."
end
upload
end
它似乎适用于我从 XenForo 2.3 到 Discourse 的导入。
结果仍然不理想。在导入约300万到400万条数据后出现错误，并提示缺少 XFMG。
我得再鼓足干劲，才能让它正常运行。
我重构了大量的代码。目前看起来运行得相当不错。你可能拥有的自定义字段不会被导入，但我又重新编写了这部分代码（见下文），并会清理我的导入数据重新开始。
已确认在我的 Xenforo 2.3 到自托管 Discourse 的迁移中正常工作：
用户
自定义用户字段
私信 / 会话
XenForo 媒体库
Xenforo 资源管理器
点赞和反应
重复和无效电子邮件的处理
XF2.3 BB 代码处理
XF 媒体库
XF 资源
我在一拥有 75 万帖子、2 万成员和 20.7GB 附件的数据集上测试了此脚本。导入大约需要 20 小时。导入后还需要几个小时让 sidekiq 完成处理。
注意事项。
我不得不允许我的 docker 在 CSF 中访问本地数据库。
挂载路径很重要：
ATTACHMENT_DIR = ENV["XF_ATTACHMENT_DIR"] || "/shared/import/internal_data/attachments"
AVATAR_DIR = ENV["XF_AVATAR_DIR"] || "/shared/import/data/avatars"
这样可以避免你将数据移动到 docker 实例中。你是在这里建立链接
# frozen_string_literal: true
# XenForo 2.3 到 Discourse 导入脚本
# 专门为 TurboRenault (dev.turborenault.co.uk) 配置
# 支持：
# - 750,000+ 帖子，使用键集分页（O(1) 快速批处理）
# - 20.7GB 附件和头像（XF 2.3 路径解析）
# - 自定义用户资料字段和自定义主题/帖子字段
# - 用户组和次要组成员资格（防冲突的 20 字符截断）
# - 私信 / 会话和多收件人（带有已删除用户回退机制）
# - XenForo 媒体库 (XFMG)
# - XenForo 资源管理器 (XFRM，具有准确的主题和更新解析)
# - 点赞和反应
# - 重复和无效电子邮件的自动回退处理
# - XF 2.3 BBCode / 格式解析器
#
# 执行：
# cd /var/discourse
# ./launcher enter turborenault
# su - discourse
# cd /var/www/discourse
# RAILS_ENV=production bundle exec ruby /shared/xenforo23.rb
require "mysql2"
require "set"
# base.rb 的健壮路径解析
base_path = File.expand_path("../base.rb", __FILE__)
base_path = "/var/www/discourse/script/import_scripts/base.rb" unless File.exist?(base_path)
require base_path
class ImportScripts::XenForo23 < ImportScripts::Base
# 预填充的数据库凭据
XENFORO_DB = ENV["XF_DB_NAME"] || "YourDB"
DB_HOST = ENV["XF_DB_HOST"] || "172.17.0.1"
DB_PORT = (ENV["XF_DB_PORT"] || 3306).to_i
DB_USER = ENV["XF_DB_USER"] || "YourUserName"
DB_PASS = ENV["XF_DB_PASS"] || "YourPassword"
TABLE_PREFIX = ENV["XF_TABLE_PREFIX"] || "xf_"
BATCH_SIZE = (ENV["XF_BATCH_SIZE"] || 2000).to_i
# 挂载路径
ATTACHMENT_DIR = ENV["XF_ATTACHMENT_DIR"] || "/shared/import/internal_data/attachments"
AVATAR_DIR = ENV["XF_AVATAR_DIR"] || "/shared/import/data/avatars"
# 功能开关
IMPORT_PM = ENV.fetch("IMPORT_PM", "true") == "true"
IMPORT_LIKES = ENV.fetch("IMPORT_LIKES", "true") == "true"
IMPORT_XFMG = ENV.fetch("IMPORT_XFMG", "true") == "true"
IMPORT_XFRM = ENV.fetch("IMPORT_XFRM", "true") == "true"
IMPORT_CUSTOM_FIELDS = ENV.fetch("IMPORT_CUSTOM_FIELDS", "true") == "true"
def initialize
super
puts "--> 正在连接到位于 #{DB_HOST}:#{DB_PORT} 的 MySQL 数据库 '#{XENFORO_DB}'，用户为 '#{DB_USER}'..."
@client = Mysql2::Client.new(
host: DB_HOST,
port: DB_PORT,
username: DB_USER,
password: DB_PASS,
database: XENFORO_DB,
symbolize_keys: true,
encoding: "utf8mb4",
reconnect: true
)
@category_mappings = {}
@group_mappings = {}
@user_custom_fields_map = {}
@seen_emails = Set.new
@missing_files_count = 0
end
def execute
optimize_discourse_for_import
import_custom_user_field_definitions if IMPORT_CUSTOM_FIELDS
import_users
import_groups
import_categories
import_posts
import_xfrm if IMPORT_XFRM
import_xfmg if IMPORT_XFMG
import_private_messages if IMPORT_PM
import_likes if IMPORT_LIKES
restore_discourse_settings
report_missing_files_summary
end
# =========================================================================
# 针对现代 Discourse 的性能调优
# =========================================================================
def optimize_discourse_for_import
puts "", "--> 正在应用 Discourse 高速导入优化..."
begin
SiteSetting.disable_emails = "all" if SiteSetting.respond_to?(:disable_emails=)
SiteSetting.process_with_fastimage = false if SiteSetting.respond_to?(:process_with_fastimage=)
SiteSetting.max_topic_title_length = 500 if SiteSetting.respond_to?(:max_topic_title_length=)
SiteSetting.min_topic_title_length = 1 if SiteSetting.respond_to?(:min_topic_title_length=)
SiteSetting.min_post_length = 1 if SiteSetting.respond_to?(:min_post_length=)
SiteSetting.allow_duplicate_topic_titles = true if SiteSetting.respond_to?(:allow_duplicate_topic_titles=)
RateLimiter.disable rescue nil
rescue StandardError => e
puts " 优化备注: #{e.message}"
end
end
def report_missing_files_summary
if @missing_files_count > 0
puts "", " *** 在导入期间未在磁盘上找到 #{@missing_files_count} 个文件 ***"
else
puts " 所有附件/头像文件均找到，状态正常。"
end
end
def log_missing_file(type, context = {})
@missing_files_count += 1
label = type.to_s.upcase
details = context.map { |k, v| " #{k}: #{v}" }.join("\n")
puts "\n [缺失 #{label}]\n#{details}"
end
def restore_discourse_settings
puts "", "--> 正在恢复默认的 Discourse 站点设置..."
begin
SiteSetting.disable_emails = "non_staff" if SiteSetting.respond_to?(:disable_emails=)
RateLimiter.enable rescue nil
rescue StandardError
# 忽略
end
puts "--> 导入成功完成！"
end
# =========================================================================
# 1. 自定义用户资料字段
# =========================================================================
def import_custom_user_field_definitions
puts "", "--> 正在导入自定义用户字段定义..."
fields = mysql_query("SELECT field_id, field_type FROM #{TABLE_PREFIX}user_field").to_a
phrases = {}
begin
phrase_rows = mysql_query("SELECT title, phrase_text FROM #{TABLE_PREFIX}phrase WHERE title LIKE 'user_field_title.%' AND language_id = 1").to_a
phrase_rows = mysql_query("SELECT title, phrase_text FROM #{TABLE_PREFIX}phrase WHERE title LIKE 'user_field_title.%'") if phrase_rows.empty?
phrase_rows.each do |p|
fid = p[:title].to_s.sub("user_field_title.", "")
phrases[fid] = p[:phrase_text]
end
rescue StandardError => e
puts " 短语备注: #{e.message}"
end
fields.each do |row|
field_id = row[:field_id]
title = phrases[field_id].presence || field_id.to_s.tr("_", " ").split.map(&:capitalize).join(" ")
desc = "自定义资料字段: #{title}"
field_type = case row[:field_type]
when "textbox", "textarea" then "text"
when "select", "radio" then "dropdown"
when "checkbox" then "confirm"
else "text"
end
uf = UserField.find_or_create_by!(name: title) do |f|
f.description = desc
f.field_type = field_type
f.editable = true
f.show_on_profile = true
f.show_on_user_card = true
end
@user_custom_fields_map[field_id] = uf.id
end
puts " 已映射 #{@user_custom_fields_map.size} 个自定义资料字段定义。"
end
def fetch_user_custom_fields(xf_user_id)
return {} unless IMPORT_CUSTOM_FIELDS && @user_custom_fields_map.present?
sql = "SELECT field_id, field_value FROM #{TABLE_PREFIX}user_field_value WHERE user_id = #{xf_user_id.to_i}"
rows = mysql_query(sql).to_a
return {} if rows.empty?
custom_fields = {}
rows.each do |row|
discourse_field_id = @user_custom_fields_map[row[:field_id]]
next unless discourse_field_id
val = row[:field_value]
if val.to_s.start_with?("a:", "s:", "i:", "b:")
begin
val = PHP.unserialize(val)
val = val.values.join(", ") if val.is_a?(Hash)
val = val.join(", ") if val.is_a?(Array)
rescue StandardError
# 保持原样
end
elsif val.to_s.start_with?("[", "{")
begin
parsed = JSON.parse(val)
val = parsed.is_a?(Array) ? parsed.join(", ") : val
rescue JSON::ParserError
# 保持原样
end
end
custom_fields["user_field_#{discourse_field_id}"] = val.to_s if val.present?
end
custom_fields
end
# =========================================================================
# 2. 用户和头像
# =========================================================================
def import_users
puts "", "--> 正在导入用户..."
total_count = mysql_query("SELECT COUNT(*) AS count FROM #{TABLE_PREFIX}user WHERE user_state = 'valid' AND is_banned = 0").to_a.first[:count]
puts " 找到 #{total_count} 个有效用户。"
last_user_id = 0
processed = 0
loop do
sql = "SELECT u.user_id, u.username, u.email, u.custom_title, u.register_date,
u.last_activity, u.user_group_id, u.is_moderator, u.is_admin, u.is_staff, u.avatar_date
FROM #{TABLE_PREFIX}user u
WHERE u.user_state = 'valid' AND u.is_banned = 0 AND u.user_id > #{last_user_id}
ORDER BY u.user_id ASC
LIMIT #{BATCH_SIZE}"
results = mysql_query(sql).to_a
break if results.empty?
last_user_id = results.last[:user_id]
next if all_records_exist?(:users, results.map { |u| u[:user_id] })
create_users(results, total: total_count, offset: processed) do |user|
username = clean_username(user[:username])
next if username.blank?
email = clean_email(user[:email], user[:user_id])
c_fields = fetch_user_custom_fields(user[:user_id])
{
id: user[:user_id],
email: email,
username: username,
title: user[:custom_title],
created_at: Time.zone.at(user[:register_date]),
last_seen_at: Time.zone.at(user[:last_activity]),
moderator: user[:is_moderator] == 1 || user[:is_staff] == 1,
admin: user[:is_admin] == 1,
custom_fields: c_fields,
post_create_action: proc { |u| import_avatar(user[:user_id], user[:avatar_date], u) }
}
end
processed += results.size
end
end
def clean_username(name)
return "" if name.blank?
name = name.tr(" ", "_").gsub(/[^a-zA-Z0-9_\-\.]/, "")
name.first(60)
end
def clean_email(email, user_id)
email_str = email.to_s.strip.downcase
# 检查标准有效性
unless email_str.present? && email_str.include?("@") && email_str.match?(/\A[^@\s]+@[^@\s]+\.[^@\s]+\z/)
return "user_#{user_id}@imported.invalid"
end
# 检查数据集中或现有 Discourse UserEmail 表中的重复项
if @seen_emails.include?(email_str) || UserEmail.exists?(email: email_str)
prefix = email_str.split("@").first.gsub(/[^a-zA-Z0-9]/, "")
return "duplicate_#{user_id}_#{prefix}@imported.invalid"
end
@seen_emails.add(email_str)
email_str
end
def import_avatar(xf_user_id, avatar_date, imported_user)
return if avatar_date.to_i == 0
group_id = xf_user_id / 1000
filename = File.join(AVATAR_DIR, "l", group_id.to_s, "#{xf_user_id}.jpg")
filename = File.join(AVATAR_DIR, "o", group_id.to_s, "#{xf_user_id}.jpg") unless File.exist?(filename)
unless File.exist?(filename)
log_missing_file(:avatar,
user_id: xf_user_id,
expected_l: File.join(AVATAR_DIR, "l", group_id.to_s, "#{xf_user_id}.jpg"),
expected_o: File.join(AVATAR_DIR, "o", group_id.to_s, "#{xf_user_id}.jpg")
)
return
end
upload = create_upload(imported_user.id, filename, "avatar_#{xf_user_id}.jpg")
return if !upload&.persisted?
imported_user.create_user_avatar if imported_user.user_avatar.nil?
imported_user.user_avatar.update(custom_upload_id: upload.id)
imported_user.update(uploaded_avatar_id: upload.id)
rescue StandardError => e
STDERR.puts "导入用户 #{xf_user_id} 的头像时出错: #{e.message}"
end
# =========================================================================
# 3. 用户组和成员资格（防冲突截断）
# =========================================================================
def import_groups
puts "", "--> 正在导入用户组和成员资格..."
existing_group_names = Set.new(Group.pluck(:name))
xf_groups = mysql_query("SELECT user_group_id, title FROM #{TABLE_PREFIX}user_group").to_a
xf_groups.each do |g|
xf_gid = g[:user_group_id]
raw_title = CGI.unescapeHTML(g[:title].to_s.strip)
# 跳过默认的 XF 系统组（1 = 未注册，2 = 已注册）
next if [1, 2].include?(xf_gid)
sanitised_name = generate_unique_group_name(raw_title, xf_gid, existing_group_names)
next if sanitised_name.blank?
group = Group.find_by(name: sanitised_name) || Group.create!(
name: sanitised_name,
full_name: raw_title,
visibility_level: Group.visibility_levels[:members]
)
@group_mappings[xf_gid] = group.id
end
puts " 已映射 #{@group_mappings.size} 个自定义用户组。"
last_user_id = 0
loop do
relations = mysql_query(
"SELECT user_id, user_group_id
FROM #{TABLE_PREFIX}user_group_relation
WHERE user_id > #{last_user_id}
ORDER BY user_id ASC
LIMIT #{BATCH_SIZE}"
).to_a
break if relations.empty?
last_user_id = relations.last[:user_id]
relations.each do |r|
discourse_group_id = @group_mappings[r[:user_group_id]]
next unless discourse_group_id
discourse_user_id = user_id_from_imported_user_id(r[:user_id])
next unless discourse_user_id
GroupUser.find_or_create_by!(
group_id: discourse_group_id,
user_id: discourse_user_id
)
end
end
puts " 成功导入组成员资格。"
end
def generate_unique_group_name(title, group_id, existing_names)
base_slug = title.to_s.parameterize(separator: '_')
base_slug = "group_#{group_id}" if base_slug.blank?
candidate = base_slug.slice(0, 20).chomp('_')
counter = 1
while existing_names.include?(candidate)
suffix = "_#{counter}"
max_base_len = 20 - suffix.length
trimmed_base = base_slug.slice(0, max_base_len).chomp('_')
candidate = "#{trimmed_base}#{suffix}"
counter += 1
end
existing_names.add(candidate)
candidate
end
# =========================================================================
# 4. 类别（论坛层级结构）
# =========================================================================
def import_categories
puts "", "--> 正在导入类别..."
nodes = mysql_query(
"SELECT n.node_id, n.title, n.description, n.parent_node_id, n.node_name, n.display_order, n.node_type_id
FROM #{TABLE_PREFIX}node n
WHERE n.node_type_id IN ('Category', 'Forum')
ORDER BY n.parent_node_id ASC, n.display_order ASC"
).to_a
top_level = nodes.select { |n| n[:parent_node_id] == 0 }
create_categories(top_level) do |c|
{
id: c[:node_id],
name: CGI.unescapeHTML(c[:title]),
description: c[:description],
position: c[:display_order],
post_create_action: proc do |cat|
Permalink.find_or_create_by(url: "forums/#{c[:node_name]}.#{c[:node_id]}", category_id: cat.id) if c[:node_name].present?
end
}
end
top_ids = Set.new(top_level.map { |c| c[:node_id] })
second_level = nodes.select { |n| top_ids.include?(n[:parent_node_id]) }
create_categories(second_level) do |c|
{
id: c[:node_id],
name: CGI.unescapeHTML(c[:title]),
description: c[:description],
position: c[:display_order],
parent_category_id: category_id_from_imported_category_id(c[:parent_node_id]),
post_create_action: proc do |cat|
Permalink.find_or_create_by(url: "forums/#{c[:node_name]}.#{c[:node_id]}", category_id: cat.id) if c[:node_name].present?
end
}
end
second_ids = Set.new(second_level.map { |c| c[:node_id] })
deeper = nodes.reject { |n| n[:parent_node_id] == 0 || top_ids.include?(n[:node_id]) || second_ids.include?(n[:node_id]) }
deeper.each do |c|
parent = c
while parent && !second_ids.include?(parent[:node_id]) && !top_ids.include?(parent[:node_id])
parent = nodes.find { |n| n[:node_id] == parent[:parent_node_id] }
end
if parent
@category_mappings[c[:node_id]] = category_id_from_imported_category_id(parent[:node_id])
end
end
end
# =========================================================================
# 5. 主题和帖子（用于 750,000+ 帖子的实时计数器）
# =========================================================================
def import_posts
puts "", "--> 正在导入主题和帖子（750k 键集分页）..."
total_posts = mysql_query("SELECT COUNT(*) AS count FROM #{TABLE_PREFIX}post p INNER JOIN #{TABLE_PREFIX}thread t ON p.thread_id = t.thread_id WHERE p.message_state = 'visible' AND t.discussion_state = 'visible'").to_a.first[:count]
puts " 待导入帖子总数: #{total_posts}"
last_post_id = 0
processed = 0
loop do
results = mysql_query(
"SELECT p.post_id AS id,
p.thread_id AS topic_id,
t.node_id AS category_id,
t.title AS title,
t.first_post_id AS first_post_id,
t.view_count AS view_count,
t.prefix_id AS prefix_id,
p.user_id AS user_id,
p.message AS raw,
p.post_date AS created_at
FROM #{TABLE_PREFIX}post p
INNER JOIN #{TABLE_PREFIX}thread t ON p.thread_id = t.thread_id
WHERE p.message_state = 'visible'
AND t.discussion_state = 'visible'
AND p.post_id > #{last_post_id}
ORDER BY p.post_id ASC
LIMIT #{BATCH_SIZE}"
).to_a
break if results.empty?
last_post_id = results.last[:id]
next if all_records_exist?(:posts, results.map { |r| r[:id] })
create_posts(results, total: total_posts, offset: processed) do |m|
skip = false
mapped = {}
mapped[:id] = m[:id]
mapped[:user_id] = user_id_from_imported_user_id(m[:user_id]) || Discourse::SYSTEM_USER_ID
mapped[:raw] = process_xenforo_post(m[:raw], m[:id])
mapped[:created_at] = Time.zone.at(m[:created_at])
if m[:id] == m[:first_post_id]
cat_id = category_id_from_imported_category_id(m[:category_id]) || @category_mappings[m[:category_id]] || SiteSetting.uncategorized_category_id
mapped[:category] = cat_id
mapped[:title] = CGI.unescapeHTML(m[:title])
mapped[:views] = m[:view_count]
tags = fetch_thread_tags(m[:topic_id])
mapped[:tags] = tags if tags.present?
mapped[:post_create_action] = proc do |pp|
Permalink.find_or_create_by(url: "threads/#{m[:topic_id]}", topic_id: pp.topic_id)
end
else
parent = topic_lookup_from_imported_post_id(m[:first_post_id])
if parent
mapped[:topic_id] = parent[:topic_id]
else
skip = true
end
end
skip ? nil : mapped
end
processed += results.size
end
end
def fetch_thread_tags(thread_id)
sql = "SELECT t.tag FROM #{TABLE_PREFIX}tag_content tc INNER JOIN #{TABLE_PREFIX}tag t ON tc.tag_id = t.tag_id WHERE tc.content_type = 'thread' AND tc.content_id = #{thread_id.to_i}"
rows = mysql_query(sql).to_a
rows.map { |r| DiscourseTagging.clean_tag(r[:tag]) }.compact
end
# =========================================================================
# 6. XenForo 资源管理器 (XFRM)
# =========================================================================
# =========================================================================
# 6. XenForo 资源管理器 (XFRM)
# =========================================================================
def import_xfrm
puts "", "--> 正在导入 XenForo 资源管理器 (XFRM)..."
has_xfrm = mysql_query("SHOW TABLES LIKE '#{TABLE_PREFIX}rm_resource'").to_a.present?
unless has_xfrm
puts " 未找到 XFRM 表，跳过。"
return
end
xfrm_cat = Category.find_by_name("Resources") || Category.create!(name: "Resources", user_id: Discourse::SYSTEM_USER_ID)
resources = mysql_query(
"SELECT r.resource_id, r.title, r.tag_line, r.user_id, r.resource_category_id, r.resource_date, r.discussion_thread_id,
t.first_post_id, u.message AS update_message, p.message AS post_message
FROM #{TABLE_PREFIX}rm_resource r
LEFT JOIN #{TABLE_PREFIX}thread t ON t.thread_id = r.discussion_thread_id
LEFT JOIN #{TABLE_PREFIX}post p ON p.post_id = t.first_post_id
LEFT JOIN #{TABLE_PREFIX}rm_resource_update u ON u.resource_update_id = r.description_update_id
WHERE r.resource_state = 'visible'"
).to_a
puts " 找到 #{resources.size} 个 XFRM 资源。"
resources.each do |res|
raw_msg = res[:update_message].presence || res[:post_message].presence
body_text = raw_msg.present? ? process_xenforo_post(raw_msg, res[:first_post_id].to_i) : ""
tagline = res[:tag_line].present? ? CGI.unescapeHTML(res[:tag_line]) : ""
raw = ""
raw += "**#{tagline}**\n\n" if tagline.present?
raw += body_text.present? ? body_text : "*(未提供描述)*"
# --- 新增：获取实际资源文件（ZIP, PDF） ---
# 将 'v.version_id' 更新为 XenForo 2 的 'v.resource_version_id'
version_files_sql = "
SELECT a.attachment_id, a.data_id, d.filename, d.file_hash, d.user_id
FROM #{TABLE_PREFIX}attachment a
INNER JOIN #{TABLE_PREFIX}attachment_data d ON a.data_id = d.data_id
INNER JOIN #{TABLE_PREFIX}rm_resource_version v ON v.resource_version_id = a.content_id
WHERE a.content_type = 'resource_version' AND v.resource_id = #{res[:resource_id]}
"
resource_files = mysql_query(version_files_sql).to_a
if resource_files.any?
raw += "\n\n### 可下载文件\n"
resource_files.each do |file|
upload = import_xf_attachment_file(file[:data_id], file[:file_hash], file[:user_id], file[:filename])
if upload&.persisted?
html = @uploader.html_for_upload(upload, file[:filename])
raw += "\n* #{html}"
end
end
end
version_urls_sql = "
SELECT version_string, download_url
FROM #{TABLE_PREFIX}rm_resource_version
WHERE resource_id = #{res[:resource_id]} AND download_url != ''
"
external_urls = mysql_query(version_urls_sql).to_a
if external_urls.any?
raw += "\n\n### 外部下载\n"
external_urls.each do |ext|
raw += "\n* [版本 #{ext[:version_string]}](#{ext[:download_url]})"
end
end
# --- 结束新增文件获取 ---
if res[:discussion_thread_id].to_i > 0 && res[:first_post_id].to_i > 0
topic_map = topic_lookup_from_imported_post_id(res[:first_post_id].to_i)
if topic_map
raw += "\n\n---\n*原始讨论主题: [点击此处查看](/t/-/#{topic_map[:topic_id]})*"
end
end
user_id = user_id_from_imported_user_id(res[:user_id]) || Discourse::SYSTEM_USER_ID
create_post = {
id: "xfrm_v5_#{res[:resource_id]}", # 升级到 v5 以清除崩溃的缓存
user_id: user_id,
category: xfrm_cat.id,
title: CGI.unescapeHTML(res[:title]),
raw: raw,
created_at: Time.zone.at(res[:resource_date])
}
create_posts([create_post], total: resources.size, offset: 0) { |p| p }
end
end
# =========================================================================
# 7. XenForo 媒体库 (XFMG)
# =========================================================================
def import_xfmg
puts "", "--> 正在导入 XenForo 媒体库 (XFMG)..."
has_xfmg = mysql_query("SHOW TABLES LIKE '#{TABLE_PREFIX}mg_media_item'").to_a.present?
unless has_xfmg
puts " 未找到 XFMG 表，跳过。"
return
end
xfmg_cat = Category.find_by_name("Media Gallery") || Category.create!(name: "Media Gallery", user_id: Discourse::SYSTEM_USER_ID)
items = mysql_query(
"SELECT m.media_id, m.title, m.description, m.media_type, m.media_tag, m.user_id, m.media_date
FROM #{TABLE_PREFIX}mg_media_item m
WHERE m.media_state = 'visible'"
).to_a
puts " 找到 #{items.size} 个 XFMG 媒体项。"
items.each do |item|
raw = item[:description].presence || ""
# 如果是嵌入视频（YouTube, Vimeo 等），包含嵌入 URL
if item[:media_type] == "embed" && item[:media_tag].present?
raw = "#{item[:media_tag]}\n\n#{raw}"
else
# 使用 XF 2.x 'xfmg_media' content_type 处理图像附件
raw = process_xf_attachments("xfmg_media", raw, item[:media_id])
end
raw = "*(无媒体内容)*" if raw.blank?
user_id = user_id_from_imported_user_id(item[:user_id]) || Discourse::SYSTEM_USER_ID
post_data = {
id: "xfmg_#{item[:media_id]}",
user_id: user_id,
category: xfmg_cat.id,
title: CGI.unescapeHTML(item[:title].presence || "Media #{item[:media_id]}"),
raw: raw,
created_at: Time.zone.at(item[:media_date])
}
create_posts([post_data], total: items.size, offset: 0) { |p| p }
end
end
# =========================================================================
# 8. 私信（会话）
# =========================================================================
def import_private_messages
puts "", "--> 正在导入私信（会话）..."
last_conv_id = 0
total_count = mysql_query("SELECT COUNT(*) AS count FROM #{TABLE_PREFIX}conversation_master").to_a.first[:count]
puts " 找到 #{total_count} 个私信会话。"
processed = 0
loop do
posts = mysql_query(
"SELECT c.conversation_id, c.title, m.message_id, m.message, m.user_id, m.message_date, c.first_message_id
FROM #{TABLE_PREFIX}conversation_master c
INNER JOIN #{TABLE_PREFIX}conversation_message m ON m.conversation_id = c.conversation_id
WHERE c.conversation_id > #{last_conv_id}
ORDER BY c.conversation_id ASC, m.message_id ASC
LIMIT #{BATCH_SIZE}"
).to_a
break if posts.empty?
last_conv_id = posts.last[:conversation_id]
next if all_records_exist?(:posts, posts.map { |p| "pm_#{p[:message_id]}" })
create_posts(posts, total: total_count, offset: processed) do |post|
user_id = user_id_from_imported_user_id(post[:user_id]) || Discourse::SYSTEM_USER_ID
message_id = "pm_#{post[:message_id]}"
raw = process_xenforo_post(post[:message], 0)
next if raw.blank?
msg = {
id: message_id,
user_id: user_id,
raw: raw,
created_at: Time.zone.at(post[:message_date].to_i),
import_mode: true
}
if post[:message_id] == post[:first_message_id]
msg[:title] = CGI.unescapeHTML(post[:title])
msg[:archetype] = Archetype.private_message
recipients_sql = "SELECT user_id FROM #{TABLE_PREFIX}conversation_recipient WHERE conversation_id = #{post[:conversation_id].to_i}"
recip_rows = mysql_query(recipients_sql).to_a
recip_user_ids = recip_rows.map { |r| user_id_from_imported_user_id(r[:user_id]) }.compact
target_usernames = User.where(id: recip_user_ids).pluck(:username)
if target_usernames.blank?
fallback_user = User.find_by(id: Discourse::SYSTEM_USER_ID) || User.admins.first
target_usernames = [fallback_user.username] if fallback_user
end
msg[:target_usernames] = target_usernames.join(",")
else
first_msg_map = topic_lookup_from_imported_post_id("pm_#{post[:first_message_id]}")
if first_msg_map
msg[:topic_id] = first_msg_map[:topic_id]
else
next
end
end
msg
end
processed += posts.size
if processed % 5000 == 0
ActiveRecord::Base.connection.clear_query_cache
GC.start
end
end
end
# =========================================================================
# 9. 反应和点赞
# =========================================================================
def import_likes
puts "", "--> 正在清除受污染的点赞并重置之前尝试的缓存..."
# 清除原始操作
PostAction.where(post_action_type_id: PostActionType.types[:like]).delete_all
# 将 UI 中缓存的计数器归零
Post.update_all(like_count: 0)
UserStat.update_all(likes_given: 0, likes_received: 0)
puts "--> 正在导入 XenForo 2.3 帖子反应（点赞）..."
# 使用 values.first 进行计数以确保跨驱动器的安全性
total_count = mysql_query(
"SELECT COUNT(*) FROM #{TABLE_PREFIX}reaction_content WHERE content_type = 'post' AND is_counted = 1"
).first.values.first
puts " 找到 #{total_count} 个活跃的帖子反应。"
last_reaction_content_id = 0
processed = 0
loop do
# 使用主键 (reaction_content_id) 安全地进行分页
reactions = mysql_query(<<-SQL).to_a
SELECT reaction_content_id, content_id, reaction_user_id, reaction_date
FROM #{TABLE_PREFIX}reaction_content
WHERE content_type = 'post'
AND is_counted = 1
AND reaction_content_id > #{last_reaction_content_id}
ORDER BY reaction_content_id ASC
LIMIT #{BATCH_SIZE}
SQL
break if reactions.empty?
# 安全地提取最高主键以用于下一个分页批次
last_row = reactions.last
last_reaction_content_id = (last_row[:reaction_content_id] || last_row["reaction_content_id"]).to_i
create_likes(reactions, total: total_count, offset: processed) do |row|
xf_post_id = row[:content_id] || row["content_id"]
xf_user_id = row[:reaction_user_id] || row["reaction_user_id"]
xf_date = row[:reaction_date] || row["reaction_date"]
# 直接传递原始的 XENFORO ID
# Discourse 的 `create_likes` 辅助方法会在后台自动转换这些 ID。
{
post_id: xf_post_id,
user_id: xf_user_id,
created_at: Time.zone.at(xf_date.to_i)
}
end
processed += reactions.size
if processed % 10000 == 0
ActiveRecord::Base.connection.clear_query_cache
GC.start
end
end
end
# =========================================================================
# 10. BBCode 解析器和格式转换器
# =========================================================================
def process_xenforo_post(raw, import_id)
return "" if raw.blank?
s = raw.dup
s = s.encode("UTF-8", invalid: :replace, undef: :replace, replace: "") unless s.valid_encoding?
s.gsub!(%r{<!-- s(\S+) --><img (?:[^>]+) /><!-- s(?:\S+) -->}, '\1')
s.gsub!(%r{<!-- \w --><a(?:.+)href="(\S+)"(?:.*)>(.+)</a><!-- \w -->}, '[\2](\1)')
s.gsub!(/:(?:\w{8})\]/, "]")
s = CGI.unescapeHTML(s)
s.gsub!(/\[QUOTE="?([^,\]]+)(?:,\s*post:\s*(\d+))?(?:,\s*member:\s*\d+)"?\),]/i) do
username = $1
imported_post_id = $2
if imported_post_id.present?
topic_mapping = topic_lookup_from_imported_post_id(imported_post_id.to_i)
if topic_mapping
"\n[quote=\"#{username}, post:#{topic_mapping[:post_number]}, topic:#{topic_mapping[:topic_id]}\"]\n"
else
"\n[quote=\"#{username}\"]\n"
end
else
"\n[quote=\"#{username}\"]\n"
end
end
s.gsub!(%r{\[/QUOTE\]}i, "\n[/quote]\n")
s.gsub!(%r{\[HEADING=1\](.+?)\[/HEADING\]}i) { "\n# #{$1}\n" }
s.gsub!(%r{\[HEADING=2\](.+?)\[/HEADING\]}i) { "\n## #{$1}\n" }
s.gsub!(%r{\[HEADING=3\](.+?)\[/HEADING\]}i) { "\n### #{$1}\n" }
s.gsub!(%r{\[SPOILER="?([^\]]*?)"?\](.*?)\[/SPOILER\]}im) do
title = $1.presence || "Spoiler"
content = $2
"\n[details=\"#{title}\"]\n#{content}\n[/details]\n"
end
s.gsub!(%r{\[CODE="?([a-zA-Z0-9_\-+]*)"?\](.*?)\[/CODE\]}im) do
lang = $1.presence || ""
code = $2
"\n```#{lang}\n#{code}\n```\n"
end
s.gsub!(%r{\[USER=\d+\]@?(.+?)\[/USER\]}i, '@\1')
s.gsub!(%r{\[MEDIA=youtube\](.+?)\[/MEDIA\]}i, 'https://www.youtube.com/watch?v=\1')
s.gsub!(%r{\[MEDIA=[^\]]+\](.+?)\[/MEDIA\]}i, '\1')
# 移除 Markdown 中不使用的表现性 BBCode 标签
s.gsub!(%r{\[/?(?:LEFT|RIGHT|CENTER|JUSTIFY|FONT|SIZE|COLOR|INDENT)(?:=[^\]]+)?\]}i, '')
# 处理 [ATTACH] 标签并将其替换为 Discourse 上传文件
s = process_xf_attachments("post", s, import_id) if import_id.to_i > 0
s.strip
end
# =========================================================================
# 11. 附件解析与文件处理
# =========================================================================
def process_xf_attachments(content_type, text, content_id)
sql = "
SELECT a.attachment_id, a.data_id, d.filename, d.file_hash, d.user_id
FROM #{TABLE_PREFIX}attachment a
INNER JOIN #{TABLE_PREFIX}attachment_data d ON a.data_id = d.data_id
WHERE a.content_type = '#{@client.escape(content_type.to_s)}'
AND a.content_id = #{content_id.to_i}
"
attachments = mysql_query(sql).to_a
return text if attachments.empty?
embedded_ids = Set.new
attachments.each do |att|
att_id = att[:attachment_id]
upload = import_xf_attachment_file(att[:data_id], att[:file_hash], att[:user_id], att[:filename])
next unless upload&.persisted?
html = @uploader.html_for_upload(upload, att[:filename])
# 匹配标准的 [ATTACH]123[/ATTACH] 以及 XF 2.3 的属性变体，如 [ATTACH type="full" alt="..."]123[/ATTACH]
pattern = %r{\[ATTACH[^\]]*\]\s*#{att_id}\s*\[/ATTACH\]}i
if text.match?(pattern)
text.gsub!(pattern, "\n#{html}\n")
embedded_ids.add(att_id)
end
end
# 追加未在正文中内联放置的附件
unattached = attachments.reject { |a| embedded_ids.include?(a[:attachment_id]) }
if unattached.any?
text += "\n\n"
unattached.each do |att|
upload = import_xf_attachment_file(att[:data_id], att[:file_hash], att[:user_id], att[:filename])
next unless upload&.persisted?
html = @uploader.html_for_upload(upload, att[:filename])
text += "#{html}\n"
end
end
# 最终清理：清除任何孤立或缺失的 [ATTACH] 标记
text.gsub!(%r{\[ATTACH[^\]]*\]\s*\d*\s*\[/ATTACH\]}i, '')
text
end
def import_xf_attachment_file(data_id, file_hash, owner_id, original_filename)
group_id = data_id.to_i / 1000
exact_name = "#{data_id}-#{file_hash}.data"
path_grouped = Pathname.new(File.join(ATTACHMENT_DIR, group_id.to_s, exact_name))
path_flat = Pathname.new(File.join(ATTACHMENT_DIR, exact_name))
path = if File.exist?(path_grouped)
path_grouped
elsif File.exist?(path_flat)
path_flat
else
glob_grouped = Dir.glob(File.join(ATTACHMENT_DIR, group_id.to_s, "#{data_id}-*.data")).first
glob_flat = Dir.glob(File.join(ATTACHMENT_DIR, "#{data_id}-*.data")).first
found = glob_grouped || glob_flat
if found
Pathname.new(found)
else
log_missing_file(:attachment,
data_id: data_id,
filename: original_filename,
db_hash: file_hash,
expected_grouped: path_grouped.to_s,
expected_flat: path_flat.to_s
)
return nil
end
end
discourse_user_id = user_id_from_imported_user_id(owner_id) || Discourse::SYSTEM_USER_ID
temp_path = path.dirname.join(original_filename)
FileUtils.cp(path, temp_path)
upload = create_upload(discourse_user_id, temp_path, original_filename)
FileUtils.rm(temp_path) if File.exist?(temp_path)
upload
rescue StandardError => e
STDERR.puts "Failed to process attachment data_id #{data_id}: #{e.message}"
nil
end
def mysql_query(sql)
@client.query(sql)
end
end
ImportScripts::XenForo23.new.perform
有些用户由于垃圾邮件拦截器（spambats）生成的奇怪电子邮件地址而导致导入失败。例如，像 Peter.Today.Rain.Import.Something@Dave.Gmail.com 这样的邮箱。
一旦自定义字段开始传输，我会更新/编辑上述导入代码，因为这些字段对我来说很重要。
.
