pip install mysqlclient --config-settings="build_options.c_options=-IC:\Program Files\MariaDB\MariaDB Connector C 64-bit\include -LC:\Program Files\MariaDB\MariaDB Connector C 64-bit\lib"
Показ дописів із міткою python. Показати всі дописи
Показ дописів із міткою python. Показати всі дописи
pip install з параметрами
Таймаут для виконання функції в python
Є скрипт під віндою, який час від часу залипає і це є проблемою. Варіант вирішення - задавати час на виконання функції, якщо він перевищує поріг, вбивати виконання функції. Нагуглив кілька варіантів, але під віндою не працює SIGALRM. А цей сніп досить гарно працює.
import multiprocessing
import time
# bar
def bar():
for i in range(100):
print "Tick"
time.sleep(1)
if __name__ == '__main__':
# Start bar as a process
p = multiprocessing.Process(target=bar)
p.start()
# Wait for 10 seconds or until process finishes
p.join(10)
# If thread is still active
if p.is_alive():
print "running... let's kill it..."
# Terminate - may not work if process is stuck for good
p.terminate()
# OR Kill - will work for sure, no chance for process to finish nicely however
# p.kill()
p.join()
Очищення imap скриньки від старих листів
Прийшло таке дивне робоче повідомлення, але здивування було більше, коли в скрині виявилось 103 тисячі повідомлень.
Ділюсь скриптом, який за 5 хвилин видалив 70 тисяч повідомлень.
#!/usr/bin/python
MAIL_SERVER = 'mail.server.com'
USERNAME = 'login@domain.com'
PASSWORD = 'pass'
MAILBOX = 'INBOX'
MAX_DAYS = 392 # Deletes messages older than a week
import imaplib
import datetime
today = datetime.date.today()
cutoff_date = today - datetime.timedelta(days=MAX_DAYS)
before_date = cutoff_date.strftime('%d-%b-%Y')
search_args = '(BEFORE "%s")' % before_date
imap = imaplib.IMAP4(MAIL_SERVER)
imap.login(USERNAME, PASSWORD)
imap.select(MAILBOX)
typ, data = imap.search(None, 'ALL', search_args)
for num in data[0].split():
print num
imap.store(num, '+FLAGS', '\\Deleted')
imap.expunge()
imap.close()
imap.logout()
pip: оновлення всіх пакунків у virtualenv
$ pip freeze --local | grep -v '^\-e' | cut -d = -f 1 | xargs -n1 pip install -Uабо
$ pip install pip-review
$ pip-review --local --interactive
Автоматичний записувач ефіру радіостанції
Виникла ідея змайструвати найпростіший повторювач (репітер, попугай) за допомогою радіостанції, комп'ютера та невеличкої схеми.
Алгоритм працю наступним чином.
З радіостанції знімається звук через вихід на колонку шнуром з резисторним дільником та включеним в мікрофонний вхід ноутбука. Резисторний дільник потрібен щоб зменшити рівень сигналу з радіостанції, бо мікрофонний вхід досить чутливий. На ноутбуці працює наступний скрипт:
і цикл, в якому відбувається запис та відтворювання записаного сигналу (останнє в стадії розробки). В гніздо навушників буде включено "дравйвер" реле, який буде слухати сигнал в навушниках і в разі його появи, реле буде спрацьовувати та імітувати натискання тангенти, також цей сигнал буде подаватись на мікрофонний вхід радіостанції. Що з того вийде, побачимо. На даний момент
і власне результат роботи скрипта:
Алгоритм працю наступним чином.
З радіостанції знімається звук через вихід на колонку шнуром з резисторним дільником та включеним в мікрофонний вхід ноутбука. Резисторний дільник потрібен щоб зменшити рівень сигналу з радіостанції, бо мікрофонний вхід досить чутливий. На ноутбуці працює наступний скрипт:
#!/bin/bash
while :
do
./rec.py
DATE=`date +%Y%m%d%H%T`
mv demo.wav demo-$DATE.wav
#vlc demo-$DATE.wav &
#killall -9 vlc
echo "recorded $DATE"
done
і цикл, в якому відбувається запис та відтворювання записаного сигналу (останнє в стадії розробки). В гніздо навушників буде включено "дравйвер" реле, який буде слухати сигнал в навушниках і в разі його появи, реле буде спрацьовувати та імітувати натискання тангенти, також цей сигнал буде подаватись на мікрофонний вхід радіостанції. Що з того вийде, побачимо. На даний момент
і власне результат роботи скрипта:
Python-скрипт для видалення старих бекапів через FTP
#!/usr/bin/python
import time
import ftputil
host = ftputil.FTPHost('ftphost', 'ftpuser', 'ftppass')
mypath = ['/', '/dir1', '/dir2', '/dir3', '/dir3',] # директорії з яких видаляти бекапи
now = time.time()
for p in mypath:
host.chdir(p)
names = host.listdir(host.curdir)
for name in names:
if host.path.getmtime(name) < (now - (7 * 86400)): # 7 днів
if host.path.isfile(name):
host.remove(name)
print 'Closing FTP connection'
host.close()
vkontakte пост на власну стіну з аплоадом фотографії на python
session = vk.Session(access_token='blahblahblah') api = vk.API(session) upload_server = api.photos.getWallUploadServer() r = requests.post(upload_server['upload_url'], files={'photo': open("/home/gsv/photo.jpg","rb")}) params = {'server': r.json()['server'], 'photo': r.json()['photo'], 'hash': r.json()['hash']} wallphoto = api.photos.saveWallPhoto(**params) photoID = wallphoto[0]['id'] params = {'attachments': str(photoID), 'message': 'Тест'} api.wall.post(**params)
Python Selenium PhantomJS робота з Facebook
По замовчуванні PhantomJS на відріз відмовився працювати з Facebook, хоча webdriver.Firefox() працює прекрасно. Методом гугління, проб і помилок вияснилось, що проблема при роботі з SSL. Отже, робочий "сніпєтс":
dcap = dict(DesiredCapabilities.PHANTOMJS) dcap["phantomjs.page.settings.userAgent"] = \ ("Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/53 (KHTML, like Gecko) Chrome/15.0.87") wd = webdriver.PhantomJS(desired_capabilities=dcap, service_args=['--ssl-protocol=tlsv1']) wd.set_window_size(1120, 550) wd.get("https://www.facebook.com/")
Прочитати змінений скриптами DOM в Selenium
Простенькі сайти можна парсити через curl+lxml чи щось інше. Динамічні сайти, які генеруються за допомогою скриптів на стороні клієнта можна парсити через Selenium і отримувати код сторінки через driver.page_source. Але є ще більш комплексні сайти (наприклад Facebook), які після завантаження і генерації сторінки додатково змінюють DOM і page_source вже не допоможе. Для вирішення питання треба запустити скрипт:
outerhtml = wd.execute_script("return document.documentElement.outerHTML")
tree = etree.parse(StringIO(outerhtml), parser)
Скрапер weblancer.net на python+lxml
Колега попросив помогти знайти помилку в скрипті. Парсалось за допомогою BS, а я його ненавиджу. Переписав на lxml
#!/usr/bin/env python
#-*- coding: utf-8 -*-
import re from lxml import etree from lxml import html as lxml_html import urllib2 from io import StringIO _base_url = "https://www.weblancer.net/projects/?page={}" def main(): parser = lxml_html.HTMLParser(encoding='windows-1251') page =urllib2.urlopen(_base_url.format(1)) data=page.read() tree = etree.parse(StringIO(data.decode("windows-1251", errors="ignore")), parser) last_page_url = tree.xpath("//ul[@class='pagination']/li[last()]/a/@href")[0] last_page = re.compile(r"(\d+)").search(last_page_url).group(1) for i in range(0, int(last_page)+1): page =urllib2.urlopen(_base_url.format(i)) data=page.read() tree = etree.parse(StringIO(data.decode("windows-1251", errors="ignore")), parser) search_results = tree.xpath("//div[@class='container-fluid cols_table show_visited']/div[@class='row']") for sr in search_results: title = sr.xpath("./div[@class='col-sm-7']/a[@class='title']/text()")[0].lstrip().rstrip() categories = sr.xpath("./div[@class='col-sm-7']/div[@class='text-muted']/a[@class='text-muted']/text()")[0].lstrip().rstrip() try: price = sr.xpath("./div[@class='col-sm-2 amount title']/text()")[0].lstrip().rstrip() except IndexError: price = ''
try: application = sr.xpath("./div[@class='col-sm-3 text-right text-nowrap hidden-xs']/text()")[0].lstrip().rstrip() except IndexError: application = '' print title, categories, application, price if __name__ == "__main__": main()
Прибирання дубляжів об'єктів моделі, які знаходяться у списку (Django, Python)
objects = [obj1, obj2, obj3,]
uniq = [] seen = set() for obj in objects: if obj.pk not in seen: uniq.append(obj) seen.add(obj.pk)
Upgrade pip from requirements.txt
pip freeze --local | grep -v '^\-e' | cut -d = -f 1 | xargs -n1 pip install -U
і ще один парсер
#!/usr/bin/env python
import re
import helpers
import hashlib
from lxml import etree
from lxml import html as lxml_html
from urlparse import urlparse
_base_url = "http://efccnigeria.org/efcc/index.php/wanted?limitstart={}"
_base_site_name = urlparse(_base_url)
def cleanhtml(raw_html):
cleanr =re.compile('<.*?>')
cleantext = re.sub(cleanr,'', raw_html)
return cleantext
def parse_links(div_blog, iter):
if iter == 0:
div_blog = div_blog[:-2]
else:
div_blog = div_blog[:-1]
links = []
for db in div_blog:
left_div = db.getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
attrib.get('href')
right_div = db.getchildren()[1].\
getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
getchildren()[0].\
attrib.get('href')
links.append("%s://%s%s" % (_base_site_name[0], _base_site_name[1], left_div))
links.append("%s://%s%s" % (_base_site_name[0], _base_site_name[1], right_div))
return links
def build_document(content):
name = content[0].getchildren()[0].text.lstrip().rstrip()
description = etree.tostring(content[2].getchildren()[0], encoding="utf-8")
description_clean = cleanhtml(description)
description_clean = re.sub('\&\#13\;', '', description_clean)
description_clean = re.sub('\xc2\xa0', '', description_clean)
description_clean = "\n".join([ll.rstrip() for ll in description_clean.splitlines() if ll.strip()]) # strip blank lines
description_clean = description_clean.split("\n")
index = description_clean.index(' <!--')
description_clean = description_clean[:index]
description_clean = ' '.join(description_clean)
entity = {
"_meta": {
"id": hashlib.sha224((re.sub("[^a-zA-Z0-9]", "", name + description_clean.decode("utf8", errors="ignore")))).hexdigest(),
"entity_type": "person"
},
"name": name,
"types": ["pep"],
"fields": [
{"name": "Description", "value": description_clean.decode("utf8", errors="ignore")}
]
}
helpers.emit(entity)
def main():
parser = lxml_html.HTMLParser(encoding='utf-8')
tree = etree.parse(_base_url.format(0), parser)
last_page_url = tree.findall("//li[@class='pagination-end']/a")[0].attrib.get('href')
rx_sequence=re.compile(r"start=([0-9]+)")
last_page = rx_sequence.search(last_page_url).group(1)
for i in range(0, int(last_page)+1, 20):
tree = etree.parse(_base_url.format(i), parser)
div_blog = tree.findall("//div[@class='blog']/div")
links = parse_links(div_blog, i)
for link in links:
tree = etree.parse(link, parser)
xpath = "/html/body/div[2]/div[2]/div/div[5]/div/div/div[2]/div[2]/div/div/div[1]"
person_info = tree.xpath(xpath)[0].getchildren()
build_document(person_info)
if __name__ == "__main__":
main()
Парсер
#!/usr/bin/env python
import re
import helpers
import hashlib
from lxml import etree
from lxml import html as lxml_html
_base_url = "http://guernseyregistry.com/article/4036/Disqualified-Directors"
def check_children(element):
if element.getchildren():
return element.getchildren()[0].text
return element.text
def build_document(member):
date_of_disqualification, \
applicant_for_disqualification, \
name_of_disqualified_director, \
period_of_disqualification, \
end_of_disqualification_period = [check_children(member[i]) for i in range(0, 5)]
entity = {
"_meta": {
"id": hashlib.sha224((re.sub("[^a-zA-Z0-9]", "", name_of_disqualified_director + date_of_disqualification))).hexdigest(),
"entity_type": "person"
},
"name": name_of_disqualified_director,
"types": ["pep"],
"fields": [
{"name": "Date of disqualification", "value": date_of_disqualification},
{"name": "Applicant for disqualification", "value": applicant_for_disqualification},
{"name": "Period of disqualification", "value": period_of_disqualification},
{"name": "End of disqualification period", "value": end_of_disqualification_period}
]
}
helpers.emit(entity)
def main():
parser = lxml_html.HTMLParser(encoding='utf-8')
tree = etree.parse(_base_url, parser)
table = tree.findall("//table[@summary='disqual directors']/tbody/tr")
for tr in table[1:]:
build_document(tr.getchildren())
if __name__ == "__main__":
main()
Очистка HTML від зайвих тегів та атрибутів
Один з напрямків моєї роботи - контент менеджмент (іншими словами, набивка сайтів інформацією). Коли на сайт потрібно закинути звичайний текст з абзацами, все просто і швидко відбувається, а коли потрібно копіпастнути якусь табличку, приміром, взяту з іншого сайту, то це займає багато часу, так як мені потрібен чистий HTML-код без зайвих тегів та атрибутів, щоб жодним чином копіпаст не впливав на дизайн мого сайту. Раніше користувався автозаміною на "пустоту", але це теж не вихід. Сьогодні мене остаточно трафили шляки і вирішив повністю автоматизувати процедуру але для командної стрічки. Є ідея зробити онлайн сервіс, яким зможуть користуватись такі самі контент менеджери як і я. Питання лише, чи дійсно комусь це може полегшити життя? Це можна оформити у вільному доступі на веб-сайті з кількома формами, галочками і пімпочками та миттєвим результатом-відфільтрованим HTML-текстом.
Тепер наглядно поясню принцип роботи скрипта, написаного на Python
Є кусок HTML:
Необхідно забрати теги: a, p, span, div, strong
і атрибути: class, width, cellspacing, cellpadding, border
Код програми:
Тепер наглядно поясню принцип роботи скрипта, написаного на Python
Є кусок HTML:
Необхідно забрати теги: a, p, span, div, strong
і атрибути: class, width, cellspacing, cellpadding, border
Код програми:
from lxml import etree
from lxml import html as lxml_html
from lxml.html import clean, fromstring, tostring, _transform_result, copy
remove_attrs = ['class', 'width', 'cellspacing', 'cellpadding', 'border']
remove_tags = ['a', 'p', 'span', 'div', 'strong']
nonempty_tags = []
class MyCleaner(clean.Cleaner):
def clean_html(self, html, parser=None):
result_type = type(html)
if isinstance(html, basestring):
doc = fromstring(html, parser=parser)
else:
doc = copy.deepcopy(html)
self(doc)
return _transform_result(result_type, doc)
cleaner = MyCleaner(remove_tags=remove_tags, page_structure=False)
def squeaky_clean(html):
parser = lxml_html.HTMLParser(encoding='utf-8')
clean_html = cleaner.clean_html(html, parser)
root = fromstring(clean_html)
context = etree.iterwalk(root)
for action, elem in context:
clean_text = elem.text and elem.text.strip(' \t\r\n')
if elem.tag in nonempty_tags and \
not (len(elem) or clean_text):
elem.getparent().remove(elem)
continue
elem.text = clean_text
for badattr in remove_attrs:
if elem.attrib.has_key(badattr):
del elem.attrib[badattr]
return tostring(root, encoding='ISO-8859-1')
html = squeaky_clean(DATA)
print html
І на виході маємо результат:
Кілька версій Django на одній системі
Стикнувся з проблемкою, коли потрібно одночасно розробляти кілька проектів під різні версії Django.
Один з варіантів вирішення: felecan.com/2011/definitive-guide-installing-django-ubuntu/
files list file for package 'xxx' is missing final newline
Сьогодні при оновленні Ubuntu отримав САБЖ помилку. Наступний вигуглений скрипт вирішує проблему.
#!/usr/bin/python
# 8th November, 2009
# update manager failed, giving me the error:
# 'files list file for package 'xxx' is missing final newline' for every package.
# some Googling revealed that this problem was due to corrupt files(s) in /var/lib/dpkg/info/
# looping though those files revealed that some did not have a final new line
# this script will resolve that problem by appending a newline to all files that are missing it
# NOTE: you will need to run this script as root, e.g. sudo python newline_fixer.py
import os
dpkg_path = '/var/lib/dpkg/info/'
paths = os.listdir(dpkg_path)
for path in paths:
path = dpkg_path + path
f = open(path, 'a+')
data = f.read()
if len(data) > 1 and data[-1:] != '\n':
f.write('\n')
print 'added newline character to:', path
f.close()
Багатопотокова проксі-чекалка на Python
#!/usr/bin/python
import Queue
import threading
import MySQLdb
import urllib2
import time
class DB:
conn = None
def connect(self):
self.conn = MySQLdb.connect(host = "x.y.z.q", user = "user", passwd = "password", db = "database")
def query(self, sql):
try:
cursor = self.conn.cursor()
cursor.execute(sql)
except (AttributeError, MySQLdb.OperationalError):
self.connect()
cursor = self.conn.cursor()
cursor.execute(sql)
return cursor
proxyList = []
global db
db = DB()
curr = db.query("SELECT url FROM proxy")
result_set = curr.fetchall ()
for row in result_set:
proxyList.append(row[0])
queue = Queue.Queue()
class ThreadUrl(threading.Thread):
def __init__(self, queue):
threading.Thread.__init__(self)
self.queue = queue
def run(self):
while True:
host = self.queue.get()
try:
proxy_handler = urllib2.ProxyHandler({'http':host})
opener = urllib2.build_opener(proxy_handler)
opener.addheaders = [('User-agent','Mozilla/5.0')]
urllib2.install_opener(opener)
req = urllib2.Request("http://www.proxylists.net/proxyjudge.php")
sock=urllib2.urlopen(req, timeout= 7)
rs = sock.read(5000)
if 'ProxyLists.Net - Proxy judge ' in rs:
print "ok %s" % host
db.query("UPDATE proxy SET checked=NOW() WHERE url = '%s' " % host)
except:
print "err %s" % host
db.query("DELETE FROM proxy WHERE url LIKE '%s' " % host)
self.queue.task_done()
def main():
for i in range(20):
t = ThreadUrl(queue)
t.setDaemon(True)
t.start()
for host in proxyList:
queue.put(host)
queue.join()
main()
аплоад файлів на Youtube через консоль
Завалялось 80 гіг відео на одному віддаленому сервері. Виявляється, все просто:
# apt-get install python-gdata python-progressbarКачаємо скрипт: http://code.google.com/p/youtube-upload/
youtube-upload --email=myemail@gmail.com --password=mypassword
--title="A.S. Mutter" --description="A.S. Mutter plays Beethoven"
--category=Music --keywords="mutter, beethoven" anne_sophie_mutter.flv
І що найцікавіше, то можна задавати маску для файлів і одною командою завантажити весь каталог файлів.
Підписатися на:
Дописи (Atom)