cecli/cecli/scrape.py at main · cecli-dev/cecli · GitHub

Repository navigation

History

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

#!/usr/bin/env python

import re

import sys

import pypandoc

from cecli import __version__, urls, utils

from cecli.dump import dump # noqa: F401

coder_user_agent = f"cecli/{__version__} +{urls.website}"

# Playwright is nice because it has a simple way to install dependencies on most

# platforms.

def check_playwright():

try:

from playwright.async_api import async_playwright # noqa: F401

from playwright.sync_api import sync_playwright # noqa: F401

has_playwright = True

except ImportError:

has_playwright = False

return has_playwright

async def check_chromium():

has_chromium = False

if check_playwright():

from playwright.async_api import async_playwright

try:

async with async_playwright() as p:

browser = await p.chromium.launch()

await browser.close()

has_chromium = True

except Exception as e:

has_chromium = False

print(f"chromium errors {e}")

return has_chromium

async def install_playwright(io):

has_playwright = check_playwright()

has_chromium = await check_chromium()

if has_playwright and has_chromium:

return True

pip_cmd = utils.get_pip_install(["cecli-dev[playwright]"])

chromium_cmd = "-m playwright install --with-deps chromium"

chromium_cmd = [sys.executable] + chromium_cmd.split()

cmds = ""

if not has_playwright:

cmds += " ".join(pip_cmd) + "\n"

if not has_chromium:

cmds += " ".join(chromium_cmd) + "\n"

text = f"""For the best web scraping, install Playwright:

{cmds}

See {urls.enable_playwright} for more info.

"""

io.tool_output(text)

if not await io.confirm_ask("Install playwright?", default="y"):

return

if not has_playwright:

success, output = utils.run_install(pip_cmd)

if not success:

io.tool_error(output)

return

success, output = utils.run_install(chromium_cmd)

if not success:

io.tool_error(output)

return

return True

class Scraper:

pandoc_available = None

playwright_available = None

playwright_instructions_shown = False

# Public API...

def __init__(self, print_error=None, playwright_available=None, verify_ssl=True):

"""

`print_error` - a function to call to print error/debug info.

`verify_ssl` - if False, disable SSL certificate verification when scraping.

"""

if print_error:

self.print_error = print_error

else:

self.print_error = print

self.playwright_available = playwright_available

self.verify_ssl = verify_ssl

async def scrape(self, url):

"""

Scrape a url and turn it into readable markdown if it's HTML.

If it's plain text or non-HTML, return it as-is.

`url` - the URL to scrape.

"""

if self.playwright_available:

content, mime_type = await self.scrape_with_playwright(url)

else:

content, mime_type = self.scrape_with_httpx(url)

if not content:

self.print_error(f"Failed to retrieve content from {url}")

return None

# Check if the content is HTML based on MIME type or content

if (mime_type and mime_type.startswith("text/html")) or (

mime_type is None and self.looks_like_html(content)

self.try_pandoc()

content = self.html_to_markdown(content)

elif mime_type and mime_type.startswith("image/"):

content = "(Response is an image, not HTML)"

return content

def looks_like_html(self, content):

"""

Check if the content looks like HTML.

"""

if isinstance(content, str):

# Check for common HTML tags

html_patterns = [

r"<!DOCTYPE\s+html",

r"<html",

r"<head",

r"<body",

r"<div",

r"<p>",

r"<a\s+href=",

]

return any(re.search(pattern, content, re.IGNORECASE) for pattern in html_patterns)

return False

# Internals...

async def scrape_with_playwright(self, url):

import playwright # noqa: F401

from playwright.async_api import Error as PlaywrightError

from playwright.async_api import TimeoutError as PlaywrightTimeoutError

from playwright.async_api import async_playwright

async with async_playwright() as p:

try:

browser = await p.chromium.launch()

except Exception as e:

self.playwright_available = False

self.print_error(str(e))

return None, None

try:

context = await browser.new_context(ignore_https_errors=not self.verify_ssl)

page = await context.new_page()

user_agent = await page.evaluate("navigator.userAgent")

user_agent = user_agent.replace("Headless", "")

user_agent = user_agent.replace("headless", "")

user_agent += " " + coder_user_agent

await page.set_extra_http_headers({"User-Agent": user_agent})

response = None

try:

response = await page.goto(url, wait_until="networkidle", timeout=5000)

except PlaywrightTimeoutError:

self.print_error(f"Page didn't quiesce, scraping content anyway: {url}")

response = None

except PlaywrightError as e:

self.print_error(f"Error navigating to {url}: {str(e)}")

return None, None

try:

content = await page.content()

mime_type = None

if response:

content_type = await response.header_value("content-type")

if content_type:

mime_type = content_type.split(";")[0]

except PlaywrightError as e:

self.print_error(f"Error retrieving page content: {str(e)}")

content = None

mime_type = None

finally:

await browser.close()

return content, mime_type

def scrape_with_httpx(self, url):

from cecli.http import httpx

headers = {"User-Agent": f"Mozilla./5.0 ({coder_user_agent})"}

try:

with httpx.Client(

headers=headers, verify=self.verify_ssl, follow_redirects=True

) as client:

response = client.get(url)

response.raise_for_status()

return response.text, response.headers.get("content-type", "").split(";")[0]

except httpx.HTTPError as http_err:

self.print_error(f"HTTP error occurred: {http_err}")

except Exception as err:

self.print_error(f"An error occurred: {err}")

return None, None

def try_pandoc(self):

if self.pandoc_available is not None:

return

try:

pypandoc.get_pandoc_version()

self.pandoc_available = True

return

except OSError:

pass

try:

pypandoc.download_pandoc(delete_installer=True)

self.pandoc_available = True

except Exception:

self.pandoc_available = False

def html_to_markdown(self, page_source):

from bs4 import BeautifulSoup

soup = BeautifulSoup(page_source, "html.parser")

soup = slimdown_html(soup)

page_source = str(soup)

if not self.pandoc_available:

return page_source

try:

md = pypandoc.convert_text(page_source, "markdown", format="html")

except OSError:

return page_source

md = re.sub(r"</div>", " ", md)

md = re.sub(r"<div>", " ", md)

md = re.sub(r"\n\s*\n", "\n\n", md)

return md

def slimdown_html(soup):

for svg in soup.find_all("svg"):

svg.decompose()

if soup.img:

soup.img.decompose()

for tag in soup.find_all(href=lambda x: x and x.startswith("data:")):

tag.decompose()

for tag in soup.find_all(src=lambda x: x and x.startswith("data:")):

tag.decompose()

for tag in soup.find_all(True):

for attr in list(tag.attrs):

if attr != "href":

tag.attrs.pop(attr, None)

return soup

async def main(url):

scraper = Scraper(playwright_available=check_playwright())

content = await scraper.scrape(url)

print(content)

if __name__ == "__main__":

if len(sys.argv) < 2:

print("Usage: python playw.py <URL>")

sys.exit(1)

main(sys.argv[1])

cecli/cecli/scrape.py at main · cecli-dev/cecli · GitHub

Repository navigation

Expand file tree

Breadcrumbs

scrape.py

Latest commit

History

Breadcrumbs

scrape.py

File metadata and controls