Is there a method with packet capture capabilities similar to Fiddler? Using this method may result in errors and carries the risk of omissions.

[yunfenglaoshi]

async with AsyncChromeDaemon(host="127.0.0.1", headless=False, debug=False, port = self.port,
                                 extra_config=[f'--user-data-dir={self.infos["path"]}', '--disable-features=PrivacySandboxSettings4'],
                                 ) as cd:
        async with cd.connect_tab(index=0, auto_close=True) as tab:
            async with tab.iter_fetch(fixed_request_patterns) as fetch:
                url = 'https://www.tiktok.com/@changhongofficial'
                export_task = asyncio.create_task(self.operate(tab,boke))
                capture_task = asyncio.create_task(self.capture(tab,fetch, fixed_request_patterns))   # This is the method for packet capture
                await tab.goto(url)
                await self.getPid(cd)
                await asyncio.gather(export_task, capture_task)

This is the method for packet capture

async def capture(self,tab,fetch,fixed_request_patterns):
    try:
        async for event in fetch:   #  There is a risk of packet loss。
            if fetch.match_event(event, fixed_request_patterns[0]):
                response = await fetch.get_response(event, timeout=5)
                itemDate = json.loads(response["data"])
                currentUrl = response['params']['response']['url']
                self.infoData = await self.parse_chunk(itemDate)
            elif fetch.match_event(event, fixed_request_patterns[1]):
                response = await fetch.get_response(event, timeout=5)
                self.commentUrl = response['params']['response']['url']
                itemDate = json.loads(response["data"])
                self.commentData = await self.parse_comments(itemDate)

    except asyncio.TimeoutError:
          pass
async def sendComment(self):
    pass

[ClericPy]

1. asyncio.create_task(event_callback) instead of fetch.get_response, or use asyncio.Queue
2. else: await f.continueRequest(data), or asyncio.create_task(f.continueRequest(data))

[yunfenglaoshi]

1. asyncio.create_task(event_callback) instead of fetch.get_response, or use asyncio.Queue
2. else: await f.continueRequest(data), or asyncio.create_task(f.continueRequest(data))
   sorry i can not stand your means ，there is some error in my works

[yunfenglaoshi]

I think I might get what you mean. Is that right?

[ClericPy]

async def network_sniffer():
    """network flow sniffer"""

    async with AsyncChromeDaemon(headless=True) as cd:
        async with cd.connect_tab(0) as tab:
            stop_sig = asyncio.Future()

            async def cb(event, tab, buffer):
                print(
                    "Event:",
                    event["method"],
                    "Tab ID:URL:",
                    event["params"].get("request", {}).get("url"),
                )
                await buffer.continueRequest(event)
                # get response body
                response = await f.get_response(event, timeout=5)
                if response:
                    print("response body:", response["data"])
                else:
                    print("No response body")
                stop_sig.set_result(True)

            async with (
                tab.iter_fetch(
                    patterns=[
                        {
                            "urlPattern": "*myip.ipip.net*",  # wildcard pattern to filter requests
                            "requestStage": "Response",  # can be "Request" or "Response", response means to intercept after response headers received
                        },
                    ],
                    callback=cb,
                ) as f
            ):
                await tab.goto("https://myip.ipip.net/", timeout=0)
                async for _ in f:
                    if not stop_sig.done():
                        continue

[yunfenglaoshi]

`import asyncio
from ichrome import AsyncChromeDaemon
from scrapy_project.spiders.config_spider.self_utils.config_spider_utils import get_port

class TiktokVideoCrawler:
def init(self,source):
self.port = get_port()
self.source = source

async def create_driver(self):
    fixed_request_patterns = [{
        "urlPattern": r'*tiktok.com/api/post/item_list/*',  # 包
        "requestStage": "Response"  # 始终捕获响应阶段
    },
        {
        "urlPattern": r'*tiktok.com/api/comment/list/*',  # 评论
        "requestStage": "Response"  # 始终捕获响应阶段
    }
    ]
    async with AsyncChromeDaemon(host="127.0.0.1", headless=False, debug=False, port = self.port) as cd:
        async with cd.connect_tab(index=0, auto_close=True) as tab:

            async def cb(event,tab,buffer):
                await buffer.continueRequest(event)
                response = await fetch.get_response(event, timeout=5)
                if response:
                    print("response body:", response["data"])
                else:
                    print("No response body")
            async with(
                    tab.iter_fetch(fixed_request_patterns, callback=cb) as fetch
            ):
                capture_task = asyncio.create_task(self.p())
                await tab.goto(self.source)
                async for event in fetch:
                    pass
                await asyncio.gather(capture_task)

async def p(self):
    while 1:
        print(111)
        await asyncio.sleep(2)

def run(self):
    asyncio.run(self.create_driver())

if name == 'main':
crawler = TiktokVideoCrawler('https://www.tiktok.com/@samsungindonesia')
crawler.run()` I modified the code according to your method, but I found that it can only capture data packets once. When there are more matching data packets, it fails to capture them and instead prevents the data packets from being rendered on the page. Could you help me figure out how to fix this, please?