load id3 data into postgresql, intermediate through sqlite3 for tag filtering.

config.py

#!/usr/bin/python

conn_string = "host='localhost' dbname='radio' user='radio_queue' password='apassword'"

filldcache.py

#!/usr/bin/python

import os
from os import listdir
from os import walk
from os.path import isfile,join
import psycopg2
import re

data_config = __import__('config', globals(), locals(), [])

psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
psycopg2.extensions.register_type(psycopg2.extensions.UNICODEARRAY)

filter = re.compile('.*(mp3|ogg)',re.IGNORECASE);

f=[]
dir=os.getcwd()
print dir
dir = dir+'/'
for (dirpath,dirname,filenames) in walk(dir):
  print dirpath,dirname
  f.extend([ (fname,join(dirpath,fname)) for fname in filenames if re.match(filter,fname) ])

conn_string = data_config.conn_string
print "Connecting to database\n ->%s" % (conn_string)
conn = psycopg2.connect(conn_string)
cursor = conn.cursor()

cnt=0
for fname,fullpath in f:
  try:
    cursor.execute("select * from dcache where path_rel=%s",(fullpath,))
  except:
    print "error:",fname,fullpath
    raise
  rows = cursor.fetchall()
  if not rows:
    print fname,fullpath
    if len(fname)>128:
      print "Error inserting into song table, filename too long, skipping"
      continue
    cursor.execute("insert into song (sha256,title) values(NULL,%s) returning id_song", (fname,));
    sid = cursor.fetchone()[0]
    print "inserted id %d" % int(sid)
    cnt=cnt+1
    cursor.execute("insert into dcache (id_song,id_source,path_rel,size,status,lastused) values(%s,0,%s,0,0,NOW())", (sid,fullpath))

conn.commit()
cursor.close()
conn.close()

print cnt,"rows added."

loadid3.py

#!/usr/bin/python

import subprocess
import psycopg2
import sqlite3
import sys
import psycopg2.extensions
psycopg2.extensions.register_type(psycopg2.extensions.UNICODE)
psycopg2.extensions.register_type(psycopg2.extensions.UNICODEARRAY)

data_config = __import__('config', globals(), locals(), [])
conn_string = data_config.conn_string

def tag_sPushPostgres(cursor, sid, mp3tag, value_s):
  cursor.execute(u'insert into tag (id_song,tag_type,tag_value_i,tag_value_s,id_user, update) select %s,tag_type,0,%s,0,NOW() from mm.file_tag_map where filetag = %s', (sid,value_s,mp3tag.lower()))
  rc = cursor.rowcount
  if (rc != 1):
    sys.exit("Could not insert: %d,%s,%s\n" % (sid,mp3tag,value_s))

sqlite_conn = sqlite3.connect('test2.db')
sqlite_conn.text_factory = str
sqlite_cursor = sqlite_conn.cursor()

conn = psycopg2.connect(conn_string)
conn.set_client_encoding('UTF8') # necessary?
cursor = conn.cursor()

sqlite_cursor.execute(u"select id_song,tag,value from tag where lower(tag) in ('title','album','artist','album_artist','date') order by id_song,tag")

dfentry = sqlite_cursor.fetchone()
print repr(dfentry)
while dfentry:
  sid = dfentry[0]
  tag = dfentry[1]
  val = dfentry[2]
  try:
    ufname = unicode(val,'utf-8')
  except:
    print "Bad utf8: %d,%s,%s\n" % (sid,tag,val)

  tag_sPushPostgres(cursor,sid,tag,val)
  conn.commit()

  dfentry = sqlite_cursor.fetchone()
  print repr(dfentry)

sqlite_conn.close()

readid3.py

#!/usr/bin/python

import subprocess
import psycopg2
import sqlite3
import songutil

from songutil import tagDict

data_config = __import__('config', globals(), locals(), [])
conn_string = data_config.conn_string

def tagDict(file):
  try:
    proc = subprocess.Popen(['/usr/bin/avconv','-i',file], stdout=subprocess.PIPE,stderr=subprocess.PIPE)
  except:
    print "Error trying to get data from avconv"
    raise

  str_out,str_err = proc.communicate()

  state = 0
  res = {}

  for each in str_err.split('\n'):
    if (state == 0):
      if(re.match('\s*Metadata:\s*',each)):
        state = 1
      continue
    if (state == 1):
      if(re.match('\s*Duration:\s*',each)):
        state = 2
        continue
      lr = each.split(':',1)
      if (len(lr) > 1):
        key = lr[0].strip()
        value = lr[1].strip()
        print "%s:%s" % (key,value)
        res[key] = value

  return res

def tagPushSqlite(cursor, sid, kvp):
  for key, value in kvp.items():
    try:
      cursor.execute(u'insert into tag (id_song,tag,value) values (?,?,?)',(sid,key,value))
    except:
      cursor.execute(u'insert or replace into tag (id_song,tag,value) values (?,?,?)',(sid,key,value))


def loadSqliteDb(cursor):
  cursor.execute(u'select name from sqlite_master where type = \'table\' and name= \'tag\'')
  res = cursor.fetchall()
  if len(res) == 0:
    cursor.execute(u'create table tag (id_song integer, tag text, value text)')
    cursor.execute(u'create index tag_sid_idx on tag (id_song)')
    cursor.execute(u'create index tag_tag_idx on tag (tag collate nocase)')
    cursor.execute(u'create unique index tag_sid_tag_idx on tag (id_song,tag collate nocase)')
  else:
    cursor.execute(u'select max(id_song) from tag')
    rs = cursor.fetchall()
    if len(rs) > 0:
      return rs[0][0]
  return 0



sqlite_conn = sqlite3.connect('test2.db')
sqlite_conn.text_factory = str
sqlite_cursor = sqlite_conn.cursor()

resumesid = loadSqliteDb(sqlite_cursor)

conn = psycopg2.connect(conn_string)
cursor = conn.cursor()

print resumesid
# cursor.execute(u'select id_song,path_rel from mm.dcache where id_song >= %s order by id_song', (resumesid,))
cursor.execute(u'select id_song,path_rel from mm.dcache where id_song not in (select distinct id_song from mm.tag)')

dfentry = cursor.fetchone()
print repr(dfentry)
while dfentry:
  sid = dfentry[0]
  file = dfentry[1]
  r = tagDict(file)
  tagPushSqlite(sqlite_cursor,sid,r)
  sqlite_conn.commit()
  dfentry = cursor.fetchone()
  print repr(dfentry)

sqlite_conn.commit()
sqlite_conn.close()

conn.close()

This gist is copyright Andrew Domaszek Feb, 2013, all rights reserved.
License is BSD-new.

Execution order:

1. filldcache.py - to fill the postgresql mm.dcache table
2. readid3.py - read file paths from mm.dcache, use avconv to capture the tags, store the tags in sqlite3
3. loadid3.py - read tags from sqlite3, if it is one of the allowable tag types, push the tag into postgresql

To resume, run 2 & 3 again. To perform an incremental, run all 3.