1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
| import csv
import json
import re
import os
import sys
import time
import requests
import chardet
from datetime import datetime
from bs4 import BeautifulSoup
def detect_encoding(file_path):
"""检测文件编码"""
with open(file_path, 'rb') as f:
raw_data = f.read(1024)
result = chardet.detect(raw_data)
return result['encoding']
def extract_douban_id(url):
"""从豆瓣链接中提取数字ID"""
if not url or url.strip() == "":
return ""
url = url.split('?')[0].split('#')[0]
patterns = [
r'/subject/(\d+)/?',
r'/subject/(\d+)\.html',
r'subject/(\d+)',
]
for pattern in patterns:
match = re.search(pattern, url)
if match:
return match.group(1)
numbers = re.findall(r'\d+', url)
if numbers:
return max(numbers, key=len)
return ""
def parse_pubdate(pubdate_str):
"""解析出版日期格式"""
if not pubdate_str or pubdate_str.strip() == "":
return ""
date_str = pubdate_str.strip()
try:
if re.match(r'^[A-Za-z]{3}-\d{2}$', date_str):
month_abbr = date_str[:3]
year = int("20" + date_str[-2:])
month = datetime.strptime(month_abbr, '%b').month
return f"{year}-{month:02d}"
elif re.match(r'^\d{4}/\d{1,2}/\d{1,2}$', date_str):
date_obj = datetime.strptime(date_str, '%Y/%m/%d')
return date_obj.strftime('%Y-%m-%d')
elif re.match(r'^\d{4}-\d{1,2}-\d{1,2}$', date_str):
parts = date_str.split('-')
year, month, day = parts[0], parts[1].zfill(2), parts[2].zfill(2)
return f"{year}-{month}-{day}"
except Exception:
pass
return date_str
def get_cover_url_from_page(douban_id):
"""
从豆瓣页面直接获取封面URL
返回: (封面URL, 成功与否, 错误信息)
"""
url = f'https://book.douban.com/subject/{douban_id}/'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
}
try:
response = requests.get(url, headers=headers, timeout=10)
response.encoding = 'utf-8'
if response.status_code != 200:
return None, False, f"HTTP错误: {response.status_code}"
soup = BeautifulSoup(response.text, 'html.parser')
# 方法1: 查找meta标签 (og:image)
meta_tag = soup.find('meta', property='og:image')
if meta_tag and meta_tag.get('content'):
cover_url = meta_tag['content']
print(f"✅ 通过og:image获取封面: {cover_url[:80]}...")
return cover_url, True, None
# 方法2: 查找主图片
img_selectors = [
'#mainpic img',
'.nbg img',
'img[rel="v:photo"]',
'.subject img',
]
for selector in img_selectors:
img_tag = soup.select_one(selector)
if img_tag and img_tag.get('src'):
cover_url = img_tag['src']
print(f"✅ 通过选择器 {selector} 获取封面: {cover_url[:80]}...")
return cover_url, True, None
# 方法3: 查找所有豆瓣图片
all_imgs = soup.find_all('img', src=True)
for img in all_imgs:
src = img['src']
if 'doubanio.com' in src and ('cover' in src.lower() or 'subject' in src.lower()):
print(f"✅ 通过图片筛选获取封面: {src[:80]}...")
return src, True, None
return None, False, "未找到封面图片"
except Exception as e:
return None, False, f"获取封面URL失败: {e}"
def generate_cover_urls(douban_id):
"""
生成多个封面链接，按优先级排列
先尝试从页面获取真实封面URL，再使用备用方案
"""
urls = []
# 尝试从豆瓣页面获取真实封面URL
cover_url, success, error = get_cover_url_from_page(douban_id)
if success and cover_url:
# 如果是豆瓣图片，尝试生成不同服务器的变体
if 'doubanio.com' in cover_url:
# 提取图片ID (格式如 sXXXXXXX.jpg)
match = re.search(r'/s(\d+\.jpg)', cover_url)
if match:
img_id = match.group(1)
# 生成不同服务器的URL
servers = ['img1', 'img2', 'img3', 'img9']
for server in servers:
urls.append(f"https://{server}.doubanio.com/view/subject/m/public/s{img_id}")
# 添加原始URL
if cover_url not in urls:
urls.insert(0, cover_url)
else:
urls.append(cover_url)
else:
urls.append(cover_url)
# 如果无法获取真实URL，使用构造的URL
if not urls:
# 豆瓣官方链接（尝试常见格式）
urls.append(f"https://img9.doubanio.com/view/subject/m/public/s{douban_id}.jpg")
urls.append(f"https://img1.doubanio.com/view/subject/m/public/s{douban_id}.jpg")
# 有些封面是 s1 + 图书ID 格式
urls.append(f"https://img9.doubanio.com/view/subject/m/public/s1{douban_id}.jpg")
urls.append(f"https://img1.doubanio.com/view/subject/m/public/s1{douban_id}.jpg")
# 第三方代理服务
urls.append(f"https://images.weserv.nl/?url=img9.doubanio.com/view/subject/m/public/s{douban_id}.jpg")
urls.append(f"https://dou.img.lithub.cc/book/{douban_id}/m.jpg")
# 备用代理
urls.append(f"https://bookcover.longitood.com/bookcover/{douban_id}")
urls.append(f"https://api.xinac.net/cover/{douban_id}")
# 去重
unique_urls = []
for url in urls:
if url not in unique_urls:
unique_urls.append(url)
return unique_urls
def clean_column_names(fieldnames):
"""清理列名，移除BOM标记和空格"""
cleaned = []
for name in fieldnames:
name = name.replace('\ufeff', '').strip()
name = re.sub(r'[\u200b\u200c\u200d\u200e\u200f\ufeff]', '', name)
cleaned.append(name)
return cleaned
def fetch_book_details(book_id, max_retries=2, delay=1.5):
"""
抓取书籍详细信息：ISBN、出版社、内容简介、封面URL
返回: (isbn, publisher, intro, cover_url, success, error_message)
"""
url = f'https://book.douban.com/subject/{book_id}/'
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
}
for attempt in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=15)
response.encoding = 'utf-8'
if response.status_code != 200:
return None, None, None, None, False, f"HTTP {response.status_code}"
soup = BeautifulSoup(response.text, 'html.parser')
# 提取ISBN
isbn = None
isbn_patterns = [
r'ISBN\s*[：:]\s*([0-9X\-]+)',
r'ISBN\s*:\s*([0-9X\-]+)',
r'书号\s*[：:]\s*([0-9X\-]+)',
]
text_content = soup.get_text()
for pattern in isbn_patterns:
match = re.search(pattern, text_content, re.IGNORECASE)
if match:
isbn_candidate = match.group(1)
isbn_candidate = re.sub(r'[^\dX]', '', isbn_candidate.upper())
if 10 <= len(isbn_candidate) <= 13:
isbn = isbn_candidate
break
# 提取出版社
publisher = None
publisher_patterns = [
r'出版社\s*[：:]\s*([^\s\n]+[^\s\n,，。])',
r'出版方\s*[：:]\s*([^\s\n]+[^\s\n,，。])',
]
for pattern in publisher_patterns:
match = re.search(pattern, text_content)
if match:
publisher = match.group(1).strip()
break
# 提取内容简介
intro = ""
# 尝试多个可能的简介位置
intro_selectors = [
'div.intro',
'div#link-report',
'div.indent[itemprop="description"]',
'div.summary'
]
for selector in intro_selectors:
element = soup.select_one(selector)
if element:
# 获取所有文本，去除多余空白
intro_text = element.get_text(strip=True)
# 限制简介长度，避免太长
if len(intro_text) > 500:
intro_text = intro_text[:500] + "..."
intro = intro_text
break
# 提取封面URL
cover_url = None
# 方法1: 查找meta标签 (og:image)
meta_tag = soup.find('meta', property='og:image')
if meta_tag and meta_tag.get('content'):
cover_url = meta_tag['content']
# 方法2: 查找主图片
if not cover_url:
img_selectors = [
'#mainpic img',
'.nbg img',
'img[rel="v:photo"]',
]
for selector in img_selectors:
img_tag = soup.select_one(selector)
if img_tag and img_tag.get('src'):
cover_url = img_tag['src']
break
return isbn, publisher, intro, cover_url, True, None
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
time.sleep(delay * 2)
continue
return None, None, None, None, False, "请求超时"
except requests.exceptions.RequestException as e:
if attempt < max_retries - 1:
time.sleep(delay)
continue
return None, None, None, None, False, f"网络错误: {e}"
except Exception as e:
return None, None, None, None, False, f"解析错误: {e}"
return None, None, None, None, False, "未知错误"
def download_cover(douban_id, save_dir, max_retries=2):
"""
下载封面图片到本地目录
使用从页面获取的真实封面URL
返回: (成功, 错误信息)
"""
# 生成封面链接（现在会先尝试获取真实URL）
cover_urls = generate_cover_urls(douban_id)
if not cover_urls:
return False, "无法生成封面链接"
# 确保保存目录存在
os.makedirs(save_dir, exist_ok=True)
# 本地文件路径
local_path = os.path.join(save_dir, f"{douban_id}.jpg")
# 如果文件已存在，跳过下载
if os.path.exists(local_path):
print(f"📁 封面已存在: {local_path}")
return True, None
# 完整的浏览器请求头，模拟真实浏览器
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
'Accept-Encoding': 'gzip, deflate, br',
'DNT': '1',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'Sec-Fetch-Dest': 'document',
'Sec-Fetch-Mode': 'navigate',
'Sec-Fetch-Site': 'none',
'Sec-Fetch-User': '?1',
'Cache-Control': 'max-age=0',
'Referer': f'https://book.douban.com/subject/{douban_id}/',
}
# 尝试不同的 User-Agent
user_agents = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:121.0) Gecko/20100101 Firefox/121.0',
'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36',
]
for url in cover_urls:
for ua_idx, user_agent in enumerate(user_agents):
for attempt in range(max_retries):
try:
# 更新 User-Agent
current_headers = headers.copy()
current_headers['User-Agent'] = user_agent
# 如果是第三方代理，可能需要不同的 Referer
if 'weserv.nl' in url or 'lithub.cc' in url or 'longitood.com' in url or 'xinac.net' in url:
current_headers['Referer'] = 'https://douban.com/'
print(f" 尝试下载封面 (User-Agent {ua_idx+1}, 尝试 {attempt+1}/{max_retries}): {url[:60]}...")
response = requests.get(url, headers=current_headers, timeout=15, stream=True)
if response.status_code == 200:
# 检查是否为图片
content_type = response.headers.get('content-type', '').lower()
if 'image' in content_type or 'jpg' in content_type or 'jpeg' in content_type or 'png' in content_type:
# 保存图片
with open(local_path, 'wb') as f:
for chunk in response.iter_content(chunk_size=8192):
if chunk:
f.write(chunk)
# 检查文件大小
file_size = os.path.getsize(local_path)
if file_size > 1024: # 至少1KB
print(f"✅ 封面下载成功: {local_path} ({file_size} 字节)")
return True, None
else:
os.remove(local_path) # 删除无效文件
print(f"⚠️ 下载的封面太小 ({file_size} 字节)，可能无效")
else:
print(f" 状态码: {response.status_code}")
# 如果是403、404、418等错误，跳过这个链接
if response.status_code in [403, 404, 418]:
break # 跳出当前URL的尝试
except requests.exceptions.RequestException as e:
print(f" 请求失败: {e}")
except Exception as e:
print(f" 下载失败: {e}")
# 重试前稍作延迟
if attempt < max_retries - 1:
time.sleep(1)
# 更换User-Agent前稍作延迟
time.sleep(0.5)
return False, "所有封面链接都下载失败"
def load_existing_books(json_path):
"""加载现有的书籍数据"""
existing_books = {}
if os.path.exists(json_path):
try:
with open(json_path, 'r', encoding='utf-8') as f:
books = json.load(f)
for book in books:
book_id = book.get('id')
if book_id:
existing_books[book_id] = book
print(f"📖 加载了 {len(existing_books)} 本现有书籍数据")
except Exception as e:
print(f"⚠️ 加载现有数据失败: {e}")
else:
print("📝 没有找到现有数据，将创建新文件")
return existing_books
def process_book_row(row, row_num, fetch_details=True, delay=1.5, existing_book=None, cover_dir=None):
"""处理单行书籍数据"""
# 提取书名
title_keys = ['书名', '图书名称', '标题', 'title', 'Title', '书籍名称']
title = ""
for key in title_keys:
if key in row and row[key] and str(row[key]).strip():
title = str(row[key]).strip()
break
if not title:
print(f"⚠️ 第{row_num}行: 未找到书名，跳过此行")
return None
# 提取作者信息
author_keys = ['作者', '作者/译者', 'author', 'Author', '著者']
author = ""
for key in author_keys:
if key in row and row[key]:
author = str(row[key]).strip()
break
# 提取链接
link_keys = ['条目链接', '链接', 'url', 'URL', '地址', 'douban_url']
link = ""
for key in link_keys:
if key in row and row[key]:
link = str(row[key]).strip()
break
# 提取出版日期
pubdate_keys = ['出版日期', '日期', 'pubdate', 'PubDate', '出版时间']
pubdate_raw = ""
for key in pubdate_keys:
if key in row and row[key]:
pubdate_raw = str(row[key]).strip()
break
# 提取评分
rating_keys = ['个人评分', '评分', 'rating', 'Rating', '我的评分', '打分']
rating_str = "0"
for key in rating_keys:
if key in row and row[key]:
rating_str = str(row[key]).strip()
break
# 提取短评
comment_keys = ['我的短评', '短评', 'comment', 'Comment', '评价', '评语']
comment = ""
for key in comment_keys:
if key in row and row[key]:
comment = str(row[key]).strip()
break
# 提取豆瓣ID
douban_id = extract_douban_id(link)
if not douban_id:
print(f"⚠️ 第{row_num}行《{title}》: 无法提取豆瓣ID，跳过")
return None
# 处理评分
try:
rating_5 = float(rating_str) if rating_str else 0
rating_10 = rating_5 * 2
except ValueError:
print(f"⚠️ 第{row_num}行《{title}》: 评分格式错误，使用0分")
rating_5 = 0
rating_10 = 0
# 处理出版日期
pubdate = parse_pubdate(pubdate_raw)
# 下载封面图片
cover_urls = []
if cover_dir:
print(f"🖼️ 下载封面图片...")
success, error = download_cover(douban_id, cover_dir)
if success:
# 重新生成封面URL列表，用于网页显示
cover_urls = generate_cover_urls(douban_id)
else:
print(f"⚠️ 封面下载失败: {error}")
else:
# 如果没有指定封面目录，只生成封面URL列表
cover_urls = generate_cover_urls(douban_id)
# 如果有现有书籍数据，使用现有数据
if existing_book:
book = existing_book.copy()
# 更新基础信息（这些信息可能从CSV更新）
book.update({
"title": title,
"author": author,
"pubdate": pubdate,
"rating": str(rating_10),
"rating_5": str(rating_5),
"link": link,
"comment": comment,
})
# 更新封面URL（如果有新的）
if cover_urls:
book['cover_urls'] = cover_urls
# 检查是否需要补充缺失信息
need_details = fetch_details and (not book.get('isbn') or not book.get('publisher') or not book.get('intro'))
if need_details:
print(f"🔍 补充《{title}》的缺失信息...")
isbn, publisher, intro, cover_url, success, error = fetch_book_details(douban_id, delay=delay)
if success:
if not book.get('isbn') and isbn:
book['isbn'] = isbn
if not book.get('publisher') and publisher:
book['publisher'] = publisher
if not book.get('intro') and intro:
book['intro'] = intro
# 如果有新的封面URL，也更新
if cover_url and cover_url not in book.get('cover_urls', []):
if 'cover_urls' not in book:
book['cover_urls'] = []
book['cover_urls'].insert(0, cover_url)
else:
print(f"⚠️ 《{title}》信息补充失败: {error}")
else:
# 新书，从头开始处理
# 初始化其他字段
isbn = ""
publisher = ""
intro = ""
# 如果需要抓取详细信息
if fetch_details:
print(f"🔍 正在获取《{title}》的详细信息...")
isbn, publisher, intro, cover_url, success, error = fetch_book_details(douban_id, delay=delay)
if success:
print(f"✅ 《{title}》信息获取成功")
if isbn:
print(f" ISBN: {isbn}")
if publisher:
print(f" 出版社: {publisher}")
# 如果有从页面获取的封面URL，优先使用
if cover_url and cover_url not in cover_urls:
cover_urls.insert(0, cover_url)
else:
print(f"⚠️ 《{title}》信息获取失败: {error}")
# 构建书籍对象（包含所有必要字段）
book = {
"id": douban_id,
"title": title,
"author": author,
"pubdate": pubdate,
"rating": str(rating_10), # 10分制
"rating_5": str(rating_5), # 5分制
"link": link,
"comment": comment, # 用户短评
"publisher": publisher or "",
"isbn": isbn or "",
"intro": intro or "", # 内容简介
"cover_urls": cover_urls,
"cover_local": f"/books/{douban_id}.jpg" if cover_dir else "", # 添加本地路径
}
return book
def convert_douban_csv(csv_path, json_path, fetch_details=True, delay=1.5, merge_existing=False, cover_dir=None):
"""
将豆瓣CSV转换为Hugo JSON数据
参数:
- csv_path: CSV文件路径
- json_path: JSON输出路径
- fetch_details: 是否抓取详细信息
- delay: 请求延迟（秒）
- merge_existing: 是否合并现有数据（保留手动修改）
"""
books = []
existing_books = {}
# 如果启用合并功能，加载现有数据
if merge_existing:
existing_books = load_existing_books(json_path)
try:
# 检测文件编码
print("🔍 检测文件编码...")
encoding = detect_encoding(csv_path)
print(f"📄 检测到编码: {encoding}")
# 尝试用检测到的编码读取
encodings_to_try = [encoding, 'utf-8-sig', 'gbk', 'gb2312', 'gb18030', 'big5', 'latin1']
for enc in encodings_to_try:
try:
with open(csv_path, 'r', encoding=enc) as f:
# 读取前几行来检测分隔符
sample = f.read(2048)
f.seek(0)
# 检测分隔符
if '\t' in sample:
delimiter = '\t'
print("📊 检测到制表符分隔的CSV文件")
elif ',' in sample:
delimiter = ','
print("📊 检测到逗号分隔的CSV文件")
else:
delimiter = '\t'
print("⚠️ 警告：无法确定分隔符，使用默认制表符")
# 创建CSV阅读器
reader = csv.reader(f, delimiter=delimiter)
# 读取第一行作为列名
headers = next(reader)
headers = clean_column_names(headers)
print(f"✅ 成功使用编码: {enc}")
print(f"📝 清理后的列名: {headers}")
# 创建字典读取器
f.seek(0)
next(f) # 跳过已读取的标题行
dict_reader = csv.DictReader(f, fieldnames=headers, delimiter=delimiter)
# 处理每一行数据
processed_ids = set() # 记录已处理的书籍ID
for row_num, row in enumerate(dict_reader, 1):
# 提取书名用于显示
title_keys = ['书名', '图书名称', '标题', 'title', 'Title', '书籍名称']
title = ""
for key in title_keys:
if key in row and row[key] and str(row[key]).strip():
title = str(row[key]).strip()
break
if not title:
print(f"⚠️ 第{row_num}行: 未找到书名，跳过此行")
continue
# 提取链接用于获取ID
link_keys = ['条目链接', '链接', 'url', 'URL', '地址', 'douban_url']
link = ""
for key in link_keys:
if key in row and row[key]:
link = str(row[key]).strip()
break
douban_id = extract_douban_id(link)
if not douban_id:
print(f"⚠️ 第{row_num}行《{title}》: 无法提取豆瓣ID，跳过")
continue
# 检查是否已存在此书籍数据
existing_book = existing_books.get(douban_id) if merge_existing else None
# 处理书籍数据
book = process_book_row(
row, row_num,
fetch_details=fetch_details,
delay=delay,
existing_book=existing_book,
cover_dir=cover_dir
)
if book:
books.append(book)
processed_ids.add(douban_id)
# 显示进度
if row_num <= 5 or row_num % 10 == 0:
print(f"✅ 第{row_num}本: 《{title[:20]}...》 - ID: {douban_id} - 评分: {book['rating_5']}/5")
# 添加请求延迟，避免被豆瓣屏蔽
if fetch_details and not merge_existing:
time.sleep(delay)
# 合并模式下：添加在CSV中不存在但在现有数据中的书籍（防止数据丢失）
if merge_existing:
for book_id, book in existing_books.items():
if book_id not in processed_ids:
print(f"📝 保留不在CSV中的书籍: 《{book.get('title', '未知')}》")
books.append(book)
break # 成功读取，跳出编码尝试循环
except UnicodeDecodeError:
print(f"❌ 编码 {enc} 解码失败，尝试下一个...")
continue
except Exception as e:
print(f"❌ 使用编码 {enc} 读取时出错: {e}")
continue
except FileNotFoundError:
print(f"❌ 错误: 找不到文件 {csv_path}")
return []
except Exception as e:
print(f"❌ 读取CSV文件时出错: {e}")
import traceback
traceback.print_exc()
return []
# 保存JSON文件
try:
os.makedirs(os.path.dirname(json_path), exist_ok=True)
with open(json_path, 'w', encoding='utf-8') as f:
json.dump(books, f, ensure_ascii=False, indent=2)
print(f"\n🎉 转换完成！共处理 {len(books)} 本书籍")
if merge_existing:
existing_count = len(existing_books)
new_books = [b for b in books if b['id'] not in existing_books]
print(f"💾 合并结果:")
print(f" - 原有书籍: {existing_count} 本")
print(f" - 新增书籍: {len(new_books)} 本")
print(f" - 保留修改: {existing_count} 本")
return books
except Exception as e:
print(f"❌ 保存JSON文件时出错: {e}")
return []
def find_csv_file(csv_file):
"""智能查找CSV文件"""
# 如果文件存在，直接返回
if os.path.exists(csv_file):
return os.path.abspath(csv_file)
# 如果文件在当前目录
if os.path.exists(os.path.join(os.getcwd(), csv_file)):
return os.path.join(os.getcwd(), csv_file)
# 如果文件在脚本所在目录
script_dir = os.path.dirname(os.path.abspath(__file__))
script_csv_path = os.path.join(script_dir, csv_file)
if os.path.exists(script_csv_path):
return script_csv_path
# 如果文件在脚本所在目录的父目录
parent_dir = os.path.dirname(script_dir)
parent_csv_path = os.path.join(parent_dir, csv_file)
if os.path.exists(parent_csv_path):
return parent_csv_path
# 尝试常见文件名
common_names = ['books.csv', '豆瓣.csv', 'douban.csv', 'booklist.csv', 'book.csv']
for name in common_names:
for base_dir in [os.getcwd(), script_dir, parent_dir]:
test_path = os.path.join(base_dir, name)
if os.path.exists(test_path):
print(f"📂 自动找到文件: {test_path}")
return test_path
return None
def main():
print("=" * 60)
print("📚 豆瓣CSV转Hugo JSON工具（带合并功能）")
print("=" * 60)
# 解析命令行参数
import argparse
parser = argparse.ArgumentParser(description='豆瓣CSV转Hugo JSON工具')
parser.add_argument('csv_file', nargs='?', default='books.csv',
help='CSV文件路径（默认: books.csv）')
parser.add_argument('--no-fetch', action='store_true',
help='不抓取详细信息（仅使用CSV中的数据）')
parser.add_argument('--merge', action='store_true',
help='合并现有数据（保留手动修改的内容）')
parser.add_argument('--delay', type=float, default=1.5,
help='请求延迟时间（秒，默认: 1.5）')
parser.add_argument('--output', '-o',
help='JSON输出路径（默认: 与CSV同目录同名）')
args = parser.parse_args()
# 智能查找CSV文件
csv_file = find_csv_file(args.csv_file)
if csv_file is None:
print(f"\n❌ 错误: 找不到文件 {args.csv_file}")
print("请检查文件是否存在，或提供完整路径。")
print("尝试查找的位置:")
print(f" 当前目录: {os.getcwd()}")
print(f" 脚本目录: {os.path.dirname(os.path.abspath(__file__))}")
print(f" 父目录: {os.path.dirname(os.path.dirname(os.path.abspath(__file__)))}")
print("\n使用方法:")
print(" 1. 使用相对路径: python scripts/douban.py ../books.csv")
print(" 2. 使用绝对路径: python scripts/douban.py /path/to/books.csv")
print(" 3. 将CSV文件放在脚本同一目录或当前目录")
sys.exit(1)
print(f"✅ 找到CSV文件: {csv_file}")
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
# 确定输出路径
if args.output:
json_file = args.output
else:
csv_name = os.path.basename(csv_file)
json_name = os.path.splitext(csv_name)[0] + '.json'
data_dir = os.path.join(PROJECT_ROOT, 'data')
json_file = os.path.join(data_dir, json_name)
os.makedirs(data_dir, exist_ok=True)
print(f"📂 CSV文件: {csv_file}")
print(f"📂 JSON输出: {json_file}")
print(f"🔍 抓取详细信息: {'否' if args.no_fetch else '是'}")
print(f"🤝 合并现有数据: {'是' if args.merge else '否'}")
if not args.no_fetch:
print(f"⏱️ 请求延迟: {args.delay}秒")
# 添加封面保存目录
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
cover_dir = os.path.join(PROJECT_ROOT, 'static', 'books')
print(f"🖼️ 封面保存目录: {cover_dir}")
# 检查依赖
try:
from bs4 import BeautifulSoup
import chardet
except ImportError:
print("\n❌ 缺少依赖，请运行: pip install requests beautifulsoup4 chardet")
sys.exit(1)
# 转换CSV为JSON
books = convert_douban_csv(
csv_file,
json_file,
fetch_details=not args.no_fetch,
delay=args.delay,
merge_existing=args.merge,
cover_dir=cover_dir
)
if books:
print(f"\n{'='*60}")
print("🎉 转换完成！")
print(f"📁 JSON文件已保存到: {os.path.abspath(json_file)}")
# 显示统计信息
has_isbn = sum(1 for b in books if b.get('isbn'))
has_publisher = sum(1 for b in books if b.get('publisher'))
has_intro = sum(1 for b in books if b.get('intro'))
has_comment = sum(1 for b in books if b.get('comment'))
has_cover = sum(1 for b in books if b.get('cover_urls') and len(b.get('cover_urls', [])) > 0)
print(f"\n📊 数据统计:")
print(f" 总书籍数: {len(books)}")
print(f" 包含ISBN: {has_isbn}")
print(f" 包含出版社: {has_publisher}")
print(f" 包含简介: {has_intro}")
print(f" 包含短评: {has_comment}")
print(f" 包含封面URL: {has_cover}")
# 显示评分分布
ratings_5 = [float(b.get('rating_5', 0)) for b in books]
avg_rating_5 = sum(ratings_5) / len(ratings_5) if ratings_5 else 0
print(f" 平均评分: {avg_rating_5:.1f}/5 ({avg_rating_5*2:.1f}/10)")
# 显示生成的JSON文件位置
print(f"\n📂 生成的JSON文件位置:")
print(f" 绝对路径: {os.path.abspath(json_file)}")
# 检查文件是否存在
if os.path.exists(json_file):
file_size = os.path.getsize(json_file)
print(f" 文件大小: {file_size} 字节")
# 显示JSON结构示例
print(f"\n📋 JSON结构示例:")
if books:
example = books[0].copy()
# 简短的示例，避免显示太多内容
if 'intro' in example and len(example['intro']) > 100:
example['intro'] = example['intro'][:100] + "..."
if 'cover_urls' in example and example['cover_urls']:
example['cover_urls'] = [example['cover_urls'][0]] # 只显示第一个
print(json.dumps(example, ensure_ascii=False, indent=2))
else:
print(" ⚠️ 警告: JSON文件未找到，可能保存失败")
# 使用建议
print(f"\n💡 使用建议:")
if args.merge:
print(" ✅ 合并模式已启用，手动修改的内容已保留")
print(" 📝 下次更新时，继续使用 --merge 参数")
else:
print(" ⚠️ 未使用合并模式，所有数据已重新生成")
print(" 💡 如需保留手动修改，下次使用 --merge 参数")
print(f"{'='*60}")
if __name__ == '__main__':
main()
|
Aria
