カクヨムのバックアップZIPから、各エピソードの本文だけを取り出して
episode_XXXX.txt としてGoogle Driveへ保存するGoogle Colab用Pythonスクリプトです。
カクヨムのバックアップZIPを取得した後の処理を自動化するために作りました。
できること
- Google DriveをColabへマウント
- 指定フォルダにある最新のZIPを自動選択
- ZIP内の
episode_XXXX.txtを抽出 -
【本文（○行）】を使って本文範囲を判定 - 本文先頭に重複したタイトル行があれば削除
- 古い
episode_XXXX.txtを削除して最新状態へ更新 -
manifest.csvを生成 - カクヨム側の
【文字数】と抽出本文の文字数が大きくずれた場合に警告
使い方
Google Colabのセルへ下記スクリプトを貼り付けて実行します。
最初に変更するのは、基本的に次の2か所です。
ZIP_SOURCE_DIR = "/content/drive/MyDrive/YOUR_PROJECT/zip"
EPISODE_OUTPUT_DIR = "/content/drive/MyDrive/YOUR_PROJECT/episodes"
ZIP_SOURCE_DIR にはカクヨムから取得したZIPの保存先、
EPISODE_OUTPUT_DIR には抽出後のエピソード保存先を指定します。
スクリプト
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
kakuyomu_body_extractor_colab.py
Google Colab上で、Google Drive内の最新ZIPから
カクヨムのエピソード本文を抽出するスクリプト。
主な処理:
- Google Driveをマウント
- ZIP保存元フォルダ内で更新日時が最新のZIPだけを対象にする
- ZIPの新旧にかかわらず、選択した最新ZIPを毎回処理する
- 保存先にある古い episode_XXXX.txt を削除
- ZIP内の episode_XXXX.txt のみ処理
- 【本文（○行）】の行数を使って本文の終端を決定
- 【文字数】と本文文字数を照合し、空白類の数え方を考慮してもTHRESHOLD以上ずれていれば警告
- 本文先頭に重複している「第○話 タイトル」行を任意で除外
- 出力先に episode_XXXX.txt として本文を書き出す
- manifest.csv にタイトル、元ファイル名、文字数、元ZIP情報などを書き出す
通常変更する設定:
- ZIP_SOURCE_DIR
- EPISODE_OUTPUT_DIR
"""
from __future__ import annotations
import csv
import re
import shutil
import zipfile
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import List, Optional, Tuple
from zoneinfo import ZoneInfo
# ============================================================
# Google Colab設定
# ============================================================
# Google Driveをマウントするか
MOUNT_GOOGLE_DRIVE = True
# Google Driveのマウント先
GOOGLE_DRIVE_MOUNT_POINT = "/content/drive"
# ============================================================
# 入出力設定
# ============================================================
# ZIPファイルが保存されているフォルダ
ZIP_SOURCE_DIR = "/content/drive/MyDrive/YOUR_PROJECT/zip"
# 抽出したエピソードファイルの保存先
EPISODE_OUTPUT_DIR = "/content/drive/MyDrive/YOUR_PROJECT/episodes"
# 対象ZIPの検索パターン
ZIP_FILE_PATTERN = "*.zip"
# manifestファイル名
MANIFEST_FILE_NAME = "manifest.csv"
# ============================================================
# 抽出設定
# ============================================================
# 本文先頭に重複しているタイトル行を削除するか
REMOVE_DUPLICATE_TITLE = True
# 処理実行時に古い episode_XXXX.txt を削除するか
DELETE_OLD_EPISODE_FILES = True
# ZIP内テキストファイルの文字コード
ZIP_TEXT_ENCODING = "utf-8"
# 【文字数】とのズレがこの割合以上なら警告
CHAR_COUNT_WARNING_THRESHOLD = 0.20
# ============================================================
# 日時設定
# ============================================================
TIMEZONE_NAME = "Asia/Tokyo"
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S%z"
@dataclass
class ExtractedEpisode:
file_name: str
number: int
title: str
body: str
source_chars: int
body_chars: int
class KakuyomuBodyExtractor:
def __init__(
self,
input_zip_path: Path,
output_dir: Path,
manifest_file_name: str,
remove_duplicate_title: bool = True,
delete_old_episode_files: bool = True,
zip_text_encoding: str = "utf-8",
timezone_name: str = "Asia/Tokyo",
datetime_format: str = "%Y-%m-%dT%H:%M:%S%z",
) -> None:
self.input_zip_path = input_zip_path
self.output_dir = output_dir
self.manifest_file_name = manifest_file_name
self.remove_duplicate_title = remove_duplicate_title
self.delete_old_episode_files = delete_old_episode_files
self.zip_text_encoding = zip_text_encoding
self.timezone = ZoneInfo(timezone_name)
self.datetime_format = datetime_format
@property
def manifest_path(self) -> Path:
return self.output_dir / self.manifest_file_name
def run(self) -> None:
self.output_dir.mkdir(parents=True, exist_ok=True)
raw_files = self.read_zip_files(self.input_zip_path)
episodes: List[ExtractedEpisode] = []
for file_name, text in raw_files:
number = self.extract_episode_number(file_name)
if number is None:
continue
title = self.extract_title(text, number)
raw_body, declared_lines = self.find_body_block(text)
raw_body = self.normalize_line_endings(raw_body).strip()
declared_chars = self.extract_declared_char_count(text)
actual_chars_min, actual_chars_max = self.count_text_chars_range(raw_body)
self.warn_if_char_count_mismatch(
declared_chars=declared_chars,
actual_chars_min=actual_chars_min,
actual_chars_max=actual_chars_max,
file_name=file_name,
)
body = raw_body
if self.remove_duplicate_title:
body = self.remove_leading_duplicate_title(body, title)
body = body.strip()
episode = ExtractedEpisode(
file_name=file_name,
number=number,
title=title,
body=body,
source_chars=len(text),
body_chars=len(body),
)
episodes.append(episode)
episodes.sort(key=lambda ep: ep.number)
if not episodes:
raise RuntimeError(
"ZIP内に episode_XXXX.txt 形式の対象ファイルがありません。"
)
if self.delete_old_episode_files:
self.delete_old_outputs()
for episode in episodes:
output_path = self.output_dir / f"episode_{episode.number:04d}.txt"
output_path.write_text(
self.build_output_text(episode),
encoding="utf-8",
)
self.write_manifest(episodes)
print(f"本文抽出完了: {self.output_dir}")
print(f"対象ZIP: {self.input_zip_path.name}")
print(f"対象話数: {len(episodes)}")
print(f"manifest: {self.manifest_path}")
def read_zip_files(self, zip_path: Path) -> List[Tuple[str, str]]:
results: List[Tuple[str, str]] = []
with zipfile.ZipFile(zip_path, "r") as zf:
for name in zf.namelist():
if not name.lower().endswith(".txt"):
continue
file_name = Path(name).name
if self.extract_episode_number(file_name) is None:
continue
data = zf.read(name)
try:
text = data.decode(self.zip_text_encoding)
except UnicodeDecodeError as exc:
raise RuntimeError(
f"文字コード {self.zip_text_encoding} で読み込めません: {name}"
) from exc
results.append((file_name, text))
return results
def extract_episode_number(self, file_name: str) -> Optional[int]:
match = re.fullmatch(r"episode_(\d{4})\.txt", file_name)
if not match:
return None
return int(match.group(1))
def extract_title(self, text: str, number: int) -> str:
match = re.search(r"【タイトル】\s*\r?\n(.+)", text)
if match:
return match.group(1).strip()
for line in text.splitlines():
stripped = line.strip()
if stripped:
return stripped
return f"第{number}話"
def extract_body(self, text: str, title: str) -> str:
body, declared_lines = self.find_body_block(text)
body = self.normalize_line_endings(body).strip()
declared_chars = self.extract_declared_char_count(text)
actual_chars_min, actual_chars_max = self.count_text_chars_range(body)
self.warn_if_char_count_mismatch(
declared_chars=declared_chars,
actual_chars_min=actual_chars_min,
actual_chars_max=actual_chars_max,
file_name=None,
)
if self.remove_duplicate_title:
body = self.remove_leading_duplicate_title(body, title)
return body.strip()
def find_body_block(self, text: str) -> Tuple[str, int]:
"""
【本文（○行）】の直後から、指定された行数だけを本文として返す。
"""
normalized = self.normalize_line_endings(text)
match = re.search(
r"(?m)^【本文（([0-9０-９]+)行）】[ \t]*\n",
normalized,
)
if not match:
raise RuntimeError(
"【本文（○行）】が見つかりません。"
"本文の終端を行数で判定できません。"
)
declared_lines = self.parse_integer(match.group(1))
remaining_lines = normalized[match.end():].splitlines()
if len(remaining_lines) < declared_lines:
raise RuntimeError(
f"【本文（{declared_lines}行）】とありますが、"
f"その後には{len(remaining_lines)}行しかありません。"
)
body = "\n".join(remaining_lines[:declared_lines])
return body, declared_lines
def extract_declared_char_count(self, text: str) -> Optional[int]:
normalized = self.normalize_line_endings(text)
patterns = [
r"【文字数】[ \t]*\n[ \t]*([0-9０-９,，]+)[ \t]*文字?",
r"【文字数】[ \t]*([0-9０-９,，]+)[ \t]*文字?",
r"【文字数（[ \t]*([0-9０-９,，]+)[ \t]*文字?[ \t]*）】",
]
for pattern in patterns:
match = re.search(pattern, normalized)
if match:
return self.parse_integer(match.group(1))
return None
def parse_integer(self, value: str) -> int:
table = str.maketrans(
"０１２３４５６７８９，",
"0123456789,",
)
return int(value.translate(table).replace(",", ""))
def count_text_chars_range(self, text: str) -> Tuple[int, int]:
"""
【文字数】との照合用に、本文文字数の妥当な範囲を返す。
actual_chars_min:
改行・半角空白・タブ・全角空白を除いた文字数
actual_chars_max:
改行だけを除いた文字数
【文字数】がこの範囲内なら、
空白類の数え方だけで説明できる差とみなす。
"""
normalized = self.normalize_line_endings(text)
actual_chars_max = len(normalized.replace("\n", ""))
without_whitespace = re.sub(r"[ \t\u3000\n]+", "", normalized)
actual_chars_min = len(without_whitespace)
return actual_chars_min, actual_chars_max
def warn_if_char_count_mismatch(
self,
declared_chars: Optional[int],
actual_chars_min: int,
actual_chars_max: int,
file_name: Optional[str],
) -> None:
prefix = f"{file_name}: " if file_name else ""
if declared_chars is None:
print(
f"WARNING: {prefix}【文字数】を取得できないため、"
"文字数照合をスキップします。"
)
return
if actual_chars_min <= declared_chars <= actual_chars_max:
return
nearest_actual = (
actual_chars_min
if declared_chars < actual_chars_min
else actual_chars_max
)
if declared_chars == 0:
diff_ratio = 0.0 if nearest_actual == 0 else 1.0
else:
diff_ratio = abs(nearest_actual - declared_chars) / declared_chars
if diff_ratio >= CHAR_COUNT_WARNING_THRESHOLD:
print(
f"WARNING: {prefix}文字数が大きくずれています。"
f"【文字数】={declared_chars}, "
f"抽出本文の妥当範囲={actual_chars_min}～{actual_chars_max}, "
f"差={diff_ratio:.1%}"
)
def normalize_line_endings(self, text: str) -> str:
return text.replace("\r\n", "\n").replace("\r", "\n")
def remove_leading_duplicate_title(self, body: str, title: str) -> str:
lines = body.splitlines()
while lines and not lines[0].strip():
lines.pop(0)
if not lines:
return ""
first = lines[0].strip()
if first == title:
return "\n".join(lines[1:]).strip()
if re.match(r"^第\d+話[ \s].+", first):
return "\n".join(lines[1:]).strip()
return body
def build_output_text(self, episode: ExtractedEpisode) -> str:
return (
"【タイトル】\n"
f"{episode.title}\n\n"
f"【本文（{len(episode.body.splitlines())}行）】\n\n"
f"{episode.body}\n"
)
def delete_old_outputs(self) -> None:
deleted_count = 0
for path in self.output_dir.iterdir():
if not path.is_file():
continue
if re.fullmatch(r"episode_\d{4}\.txt", path.name):
path.unlink()
deleted_count += 1
print(f"古いエピソードファイル削除: {deleted_count}件")
def write_manifest(self, episodes: List[ExtractedEpisode]) -> None:
source_zip_modified_at = datetime.fromtimestamp(
self.input_zip_path.stat().st_mtime,
tz=self.timezone,
)
processed_at = datetime.now(self.timezone)
fieldnames = [
"number",
"title",
"file_name",
"source_chars",
"body_chars",
"removed_chars",
"body_lines",
"source_zip_name",
"source_zip_modified_at",
"processed_at",
]
temporary_path = self.manifest_path.with_suffix(
self.manifest_path.suffix + ".tmp"
)
with temporary_path.open(
"w",
encoding="utf-8-sig",
newline="",
) as file:
writer = csv.DictWriter(file, fieldnames=fieldnames)
writer.writeheader()
for episode in episodes:
writer.writerow(
{
"number": episode.number,
"title": episode.title,
"file_name": episode.file_name,
"source_chars": episode.source_chars,
"body_chars": episode.body_chars,
"removed_chars": (
episode.source_chars - episode.body_chars
),
"body_lines": len(episode.body.splitlines()),
"source_zip_name": self.input_zip_path.name,
"source_zip_modified_at": (
source_zip_modified_at.strftime(self.datetime_format)
),
"processed_at": processed_at.strftime(
self.datetime_format
),
}
)
shutil.move(str(temporary_path), str(self.manifest_path))
def mount_google_drive() -> None:
if not MOUNT_GOOGLE_DRIVE:
return
try:
from google.colab import drive
except ImportError as exc:
raise RuntimeError(
"Google Colab環境ではありません。"
"Colab以外で実行する場合は MOUNT_GOOGLE_DRIVE = False "
"に変更してください。"
) from exc
mount_path = Path(GOOGLE_DRIVE_MOUNT_POINT)
drive.mount(str(mount_path), force_remount=False)
def find_latest_zip(
source_dir: Path,
file_pattern: str,
) -> Path:
if not source_dir.exists():
raise RuntimeError(f"ZIP保存元フォルダが存在しません: {source_dir}")
if not source_dir.is_dir():
raise RuntimeError(f"ZIP保存元がフォルダではありません: {source_dir}")
zip_files = [
path
for path in source_dir.glob(file_pattern)
if path.is_file() and path.suffix.lower() == ".zip"
]
if not zip_files:
raise RuntimeError(
f"対象ZIPが見つかりません: {source_dir / file_pattern}"
)
return max(
zip_files,
key=lambda path: (
path.stat().st_mtime,
path.name,
),
)
def main() -> None:
mount_google_drive()
zip_source_dir = Path(ZIP_SOURCE_DIR)
episode_output_dir = Path(EPISODE_OUTPUT_DIR)
latest_zip = find_latest_zip(
source_dir=zip_source_dir,
file_pattern=ZIP_FILE_PATTERN,
)
latest_zip_modified_at = datetime.fromtimestamp(
latest_zip.stat().st_mtime,
tz=ZoneInfo(TIMEZONE_NAME),
)
print(f"最新ZIP: {latest_zip.name}")
print(
"最新ZIP更新日時: "
f"{latest_zip_modified_at.strftime(DATETIME_FORMAT)}"
)
print("ZIPの新旧にかかわらず抽出処理を実行します。")
extractor = KakuyomuBodyExtractor(
input_zip_path=latest_zip,
output_dir=episode_output_dir,
manifest_file_name=MANIFEST_FILE_NAME,
remove_duplicate_title=REMOVE_DUPLICATE_TITLE,
delete_old_episode_files=DELETE_OLD_EPISODE_FILES,
zip_text_encoding=ZIP_TEXT_ENCODING,
timezone_name=TIMEZONE_NAME,
datetime_format=DATETIME_FORMAT,
)
extractor.run()
if __name__ == "__main__":
main()
補足
このスクリプトは、カクヨムのバックアップZIP内にある
episode_XXXX.txt と、ファイル内の 【タイトル】、【本文（○行）】、【文字数】
という形式を前提にしています。
サイト側のバックアップ形式が変更された場合は、修正が必要になる可能性があります。
このスクリプトを作った経緯は、カクヨムの以下の記事に書いています。
記録4：PythonスクリプトをChatGPT上で実行して小説のエピソードを分割・加工・保存した
https://kakuyomu.jp/works/2912051605768714980/episodes/2912051606082217790
この仕組みを実際に使っている作品
今回紹介した仕組みは、以下のカクヨム作品の制作・分析で実際に使用しています。
アークリーチャーズ
頭のない異形が歩く世界で、頭部を外して戦う汎用人型兵器カイを中心に描くSF小説です。
AIを使いながら、設定管理・推敲・バックアップ・読者動向の分析まで含めて制作しています。
https://kakuyomu.jp/works/822139842645600859
AIで小説を書くにはどうすればいい？
『アークリーチャーズ』をAIで制作する中で、実際に試した方法、失敗、ツールの変更、制作環境の改善などを記録しているドキュメンタリーです。