zhang_crawl_bio/demo/search_keyward.py

import asyncio
import re
from crawl4ai import *
from pathlib import Path
import json
from lxml import html  # 使用 lxml.html 模块
from sqlmodel import Session, select
from mylib.base import (replace_space, save_to_file, save_all_result,
                        save_to_pickle,ensure_output_dir,
                        save_base64_to_file,browser_config)
from mylib.drission_page import load_chrome_from_ini
from database.excel_import import ExcelDatabaseManager,KeywordModel
from config.settings import GOOGLE_SEARCH_DIR
from mylib.crawl_lib_func import filter_links,filter_local_domain

page = load_chrome_from_ini()

async def google_search(url:str, config=None)->CrawlResult:
    run_config = CrawlerRunConfig(
    magic=True,
    simulate_user=True,
    override_navigator=True
    )
    async with AsyncWebCrawler(config=browser_config) as crawler:
        result = await crawler.arun(
            url=url,
            cache_mode=CacheMode.ENABLED,
            user_agent='random',
            config=run_config,

        )
        # save_to_pickle(result, GOOGLE_SEARCH_DIR / f"{search_key}.pickle")
        return result

    
def is_search_result_empty(html_content: str) -> bool:
    '''
    检查页面是否存在 id="search" 的元素
    并检查其是否有子元素
    如果没有子元素则返回 True 表示搜索结果为空
    '''
    tree = html.fromstring(html_content)
    search_elements = tree.xpath('//*[@id="search"]/*')
    return len(search_elements) == 0

def is_already_processed(keyword: str) -> bool:
    """检查关键词是否已处理"""
    save_dir = GOOGLE_SEARCH_DIR / replace_space(keyword) / 'pkl'
    return save_dir.exists() and any(save_dir.glob("*.pickle"))

async def process_keyword(keyword: str, start=0, pages_num=250, cache=True, skip_exist=True):
    global page
    """处理单个关键词"""
    # keyword = replace_space(keyword)
    save_dir = GOOGLE_SEARCH_DIR / keyword 
    ensure_output_dir(save_dir)
    
    # # 如果已经处理过，直接返回保存目录
    # if is_already_processed(keyword):
    #     print(f"关键词 {keyword} 已处理，跳过搜索")
    #     return save_dir
        
    # 未处理过，执行搜索
    for i in range(start, pages_num, 10):
        save_html_path = GOOGLE_SEARCH_DIR / keyword / f"{i}.html"
        url = f"https://www.google.com/search?q={keyword}&start={i}"
        print(f"search url: {url}")
        
        # 如果缓存文件存在，直接读取
        # if skip_exist and save_html_path.exists():
        #     print(f"跳过缓存文件 {save_html_path}")
        #     continue
        #     print(f"读取缓存文件 {save_html_path}")
        # else:
        #     page.get(url)
        #     save_to_file(page.html,save_html_path)

        #     # result: CrawlResult = await google_search(url)
        #     # 保存 HTML 文件
        #     # save_to_file(result.html, save_html_path)
        #     print(f"保存 HTML 文件 {save_html_path}")
        # url = f"file://{save_html_path}"
        result: CrawlResult = await google_search(url)
        
        # 漂亮打印 result.links
        # print(json.dumps(result.links, indent=4))
        save_json_path = save_to_file(json.dumps(result.links, indent=4), save_dir / f"links-{i}.json")
        print(f"保存 links 文件 {save_json_path}")
        # if is_search_result_empty(result.html):
        search_res_links = filter_links(result.links)
        if not search_res_links:
            print(f"没有找到多余的页面 {result.url}，退出")
            break
            
        # links = filter_links(result.links)
        # print(f"start: {i}, links: {links} \n len: {len(links)}")
        # save_to_pickle(result, save_dir / f"result-{i}.pickle")
    return save_dir

async def search_all():
    excel_db_manager = ExcelDatabaseManager()
    """处理所有未完成的关键词"""
    key_model_list = excel_db_manager.get_keywords_by_status()
    for keyword_model in key_model_list:
        # if is_already_processed(keyword):
        #     print(f"关键词 {keyword} 已处理，跳过")
        #     continue
        await process_keyword(keyword_model.key_word)
async def test_single_search():
    # await process_keyword("Acalypha malabarica essential oil", start=0, pages_num=250)
    # save_html_path = Path(r'K:\code\upwork\zhang_crawl_bio\output\debug\正常的搜索结果.html')
    # save_html_path = Path(r'K:\code\upwork\zhang_crawl_bio\output\debug\查询不到内容.html')
    # save_html_path = Path(r"K:\code\upwork\zhang_crawl_bio\output\debug\流量异常.html")
    # save_html_path = Path(r"K:\code\upwork\zhang_crawl_bio\output\debug\最后一页.html")
    save_html_path = Path(r"K:\code\upwork\zhang_crawl_bio\output\debug\只有一页.html")
    save_dir = save_html_path.parent
    file_name = save_html_path.name
    url = f"file://{save_html_path}"
    result: CrawlResult = await google_search(url)
    # 漂亮打印 result.links
    # print(json.dumps(result.links, indent=4))
    save_json_path = save_to_file(json.dumps(result.links, indent=4), save_dir / f"links-{file_name}.json")
    print(f"保存 result.links 文件 {save_json_path}")
    links = filter_links(result.links)
    # print('\n -----------------links:')
    # print(json.dumps(links, indent=4))
    save_json_path = save_to_file(json.dumps(links, indent=4), save_dir / f"links-{file_name}-filter.json")
    links_not_local = filter_local_domain(links)
    print(f"保存 links_not_local.json 文件 {save_json_path}")
    # print('\n -----------------links_not_local:')
    # print(json.dumps(links_not_local, indent=4))
    # print(f"len links_not_local: {len(links_not_local)}")
    save_links_not_local_path = save_to_file(json.dumps(links_not_local, indent=4), save_dir / f"links-{file_name}-filter-not-local.json")
    print(f"保存 links-{file_name}-filter-not-local.json 文件 {save_links_not_local_path}")
    # print(f"start: {i}, links: {links} \n len: {len(links)}")

    # result = await google_search("Acalypha malabarica essential oil", start=50)
    # print(f"result clean html:\n {result.cleaned_html}")
    # print(f"result.links\n {result.links['external']}")
    # res = filter_links(result.links)
    # print(res)
    # 漂亮打印
    # print(json.dumps(res, indent=4))
async def test_html_to_doc():
    save_dir = Path(r"K:\code\upwork\zhang_crawl_bio\output\google_search")
    # 获取所有文件夹
    folders = [f for f in save_dir.iterdir() if f.is_dir()]
    print(folders)
    for folder in folders:
        # 获取文件夹下的所有 html 文件
        html_files = [f for f in folder.iterdir() if f.suffix == '.html']

async def test_sigle_html_links(save_html_path=None):
    if not save_html_path:
        save_html_path = Path(r"K:\code\upwork\zhang_crawl_bio\output\Acalypha malabarica essential oil\0.html")
    url = f"file://{save_html_path}"
    result: CrawlResult = await google_search(url)
    not_google_external_links = filter_links(result.links)
    links_not_local = filter_local_domain(not_google_external_links)
    # print(links_not_local)
    return links_not_local

async def test_dir_links_not_local(dir_path:Path):
    '''
    获取所有目录中的 html 文件，解析得到过滤后的域名列表， base_domain 不包含本地路径
    '''
    html_files = [f for f in dir_path.iterdir() if f.suffix == '.html']
    all_links = []
    for html_file in html_files:
        print(f"Processing {html_file}")
        links = await test_sigle_html_links(html_file)
        print(f"Found {len(links)} links in {html_file}")
        all_links.extend(links)
    print(f"Found {len(all_links)} links in total")
    # 按行写入到文件中
    save_to_file(all_links, dir_path / "links.json.txt")
    return all_links
async def main():
    # await search_all()
    # await test_single_search()
    await test_dir_links_not_local(Path(r"K:\code\upwork\zhang_crawl_bio\output\Acalypha malabarica essential oil"))
    
if __name__ == "__main__":
    asyncio.run(main())