/**

 * Created by 小陈 on 2016/3/29.

 */

@Component

public class ElasticCrawler extends BreadthCrawler {

    @Autowired

     IpaDao ipaDao;

    public ElasticCrawler() {

        super("crawl", true);

        /*start page*/

        this.addSeed("xxx");

        /*fetch url like http://news.hfut.edu.cn/show-xxxxxxhtml*/

        this.addRegex("xxx");

        /*do not fetch jpg|png|gif*/

        this.addRegex("-.*\\.(jpg|png|gif).*");

        /*do not fetch url contains #*/

//        this.addRegex("-.*#.*");

    }

    @Override

    public void visit(Page page, CrawlDatums next) {

        String url = page.getUrl();

        String content="";

        try {

             content = ContentExtractor.getContentByUrl(url);

        }catch (Exception e){

            e.printStackTrace();

        }

          /*抽取标题*/

        String title=page.getDoc().title();

        System.out.println("-------------------->"+title);

        if(!title.isEmpty() && ! content.isEmpty()){

                Pa pa=new Pa(title,content);

               ipaDao.save(pa);//持久化到数据库

            }

    }

/**

 * @author kl by 2016/3/29

 * @boke www.kailing.pub

 */

@Component

public class CSDNCrawler extends BreadthCrawler {

    @Autowired

    IpaDao ipaDao;

    public CSDNCrawler() {

        super("crawl", true);

        /*start page*/

        this.addSeed("http://blog.csdn.net/.*");//添加种子地址

        /*fetch url like http://news.hfut.edu.cn/show-xxxxxxhtml*/

        this.addRegex("http://blog.csdn.net/.*/article/details/.*");

        /*do not fetch jpg|png|gif*/

        this.addRegex("-.*\\.(jpg|png|gif).*");

        /*do not fetch url contains #*/

//        this.addRegex("-.*#.*");

    }

    @Override

    public void visit(Page page, CrawlDatums next) {

        String url = page.getUrl();

        String content="";

        try {

            content = ContentExtractor.getContentByUrl(url);

        }catch (Exception e){

            e.printStackTrace();

        }

        if (page.matchUrl("http://blog.csdn.net/.*/article/details/.*")) {

            String title = page.select("div[class=article_title]").first().text();

            String author = page.select("div[id=blog_userface]").first().text();//获取作者名

            System.out.println("title:" + title + "\tauthor:" + author);

            if(!title.isEmpty() && ! content.isEmpty()){

                Pa pa=new Pa(title,content);

                ipaDao.save(pa);

            }

        }

    }

org.elasticsearch.transport.RemoteTransportException: Failed to deserialize exception response from stream

• 广州：http://www.meetup.com/guangzhou-Elastic-Meetup/ 
• 上海：http://www.meetup.com/shanghai-Elastic-Meetup/
• 北京：http://www.meetup.com/beijing-Elastic-Meetup/

• Crate.io
• http://sqltoelasticsearch.fr/
• https://github.com/NLPchina/elasticsearch-sql

$ cat << EOF | ./es_query.py http://127.0.0.1:9200 

    WITH SELECT MAX(market_cap) AS max_all_times FROM symbol AS all_symbols; 

    WITH SELECT MAX(market_cap) AS max_at_2000 FROM all_symbols WHERE ipo_year=2000 AS year_2000; 

    WITH SELECT MAX(market_cap) AS max_at_2001 FROM all_symbols WHERE ipo_year=2001 AS year_2001;

EOF

using System;

using Nest;



namespace ConsoleApplication1

{

    class Program

    {

        private static IIndexResponse Index<T>(T person, string indexName) where T : class

        {

            var node = new Uri("http://localhost:9200&quot;);



            var settings = new ConnectionSettings(node,defaultIndex: indexName);

            var client = new ElasticClient(settings);

            return client.Index(person);

        }



        static void Main()

        {

            string indexNameError = typeof(Person).FullName

                .Substring(typeof(Person).FullName.LastIndexOf(".", StringComparison.Ordinal) + 1) + "Indexs";

            const string indexNameOk = "test";

            var person = new Person

            {

                Id = "1",

                Firstname = "Martijn",

                Lastname = "Laarman"

            };

            var ok = Index(person, indexNameOk);

            Console.WriteLine("Result:" + ok.IsValid);

            var error = Index(person, indexNameError);

            Console.WriteLine("Result:" + error.IsValid);

            Console.ReadKey();

        }

       

    }



   [Serializable]

    public class Person

    {

        public string Firstname { get; set; }

        public string Lastname { get; set; }

        public string Id { get; set; }

    }

}

// Functions to be exported by a protocol plugin

type ProtocolPlugin interface {

	// Called to initialize the Plugin

	Init(test_mode bool, results publisher.Client) error

 

	// Called to return the configured ports

	GetPorts() int

}

 

type TcpProtocolPlugin interface {

	ProtocolPlugin

 

	// Called when TCP payload data is available for parsing.

	Parse(pkt *Packet, tcptuple *common.TcpTuple,

		dir uint8, private ProtocolData) ProtocolData

 

	// Called when the FIN flag is seen in the TCP stream.

	ReceivedFin(tcptuple *common.TcpTuple, dir uint8,

		private ProtocolData) ProtocolData

 

	// Called when a packets are missing from the tcp

	// stream.

	GapInStream(tcptuple *common.TcpTuple, dir uint8, nbytes int,

		private ProtocolData) (priv ProtocolData, drop bool)

 

	// ConnectionTimeout returns the per stream connection timeout.

	// Return <=0 to set default tcp module transaction timeout.

	ConnectionTimeout() time.Duration

}

 

type UdpProtocolPlugin interface {

	ProtocolPlugin

 

	// ParseUdp is invoked when UDP payload data is available for parsing.

	ParseUdp(pkt *Packet)

}

type Packet struct {

	Ts      time.Time

	Tuple   common.IpPortTuple

	Payload byte

}

protocols:

  smtp:

    # Configure the ports where to listen for Smtp traffic. You can disable

    # the Smtp protocol by commenting out the list of ports.

    ports: [25]

git diff config/config.go

@@ -42,6 +42,7 @@ type Protocols struct {

        Pgsql    Pgsql

        Redis    Redis

        Thrift   Thrift

+       Smtp     Smtp

 }

 

 type Dns struct {

@@ -118,5 +119,9 @@ type Redis struct {

        Send_response *bool

 }

 

+type Smtp struct {

+	ProtocolCommon        `yaml:",inline"`

+}

+

 // Config Singleton

 var ConfigSingleton Config

...TODO...

git diff protos/protos.go

@@ -103,6 +103,7 @@ const (

        MongodbProtocol

        DnsProtocol

        MemcacheProtocol

+       SmtpProtocol

 )

 

 // Protocol names

@@ -116,6 +117,7 @@ var ProtocolNames = string{

        "mongodb",

        "dns",

        "memcache",

+       "smtp",

 }

git diff packetbeat.go

@@ -27,6 +27,7 @@ import (

        "github.com/elastic/packetbeat/protos/tcp"

        "github.com/elastic/packetbeat/protos/thrift"

        "github.com/elastic/packetbeat/protos/udp"

+       "github.com/elastic/packetbeat/protos/smtp"

        "github.com/elastic/packetbeat/sniffer"

 )

 

@@ -43,6 +44,7 @@ var EnabledProtocolPlugins map[protos.Protocol]protos.ProtocolPlugin = map[proto

        protos.ThriftProtocol:   new(thrift.Thrift),

        protos.MongodbProtocol:  new(mongodb.Mongodb),

        protos.DnsProtocol:      new(dns.Dns),

+       protos.SmtpProtocol:      new(smtp.Smtp),

 }

Parse(pkt *Packet, tcptuple *common.TcpTuple,

		dir uint8, private ProtocolData) ProtocolData

 priv := mysqlPrivateData{}

        if private != nil {

                var ok bool

                priv, ok = private.(mysqlPrivateData)

                if !ok {

                        priv = mysqlPrivateData{}

                }

        }

 

        [ ... ]

 

        return priv

ok, complete := mysqlMessageParser(priv.Data[dir])

                if !ok {

                        // drop this tcp stream. Will retry parsing with the next

                        // segment in it

                        priv.Data[dir] = nil

                        logp.Debug("mysql", "Ignore MySQL message. Drop tcp stream.")

                        return priv

                }

 

                if complete {

                        mysql.messageComplete(tcptuple, dir, stream)

                } else {

                        // wait for more data

                        break

                }

                

S: 220 www.example.com ESMTP Postfix

C: HELO mydomain.com

S: 250 Hello mydomain.com

C: MAIL FROM: 

S: 250 Ok

C: RCPT TO: 

S: 250 Ok

C: DATA

S: 354 End data with .

C: Subject: test message

C: From:""< sender@mydomain.com>

C: To:""< friend@example.com>

C:

C: Hello,

C: This is a test.

C: Goodbye.

C: .

S: 250 Ok: queued as 12345

C: quit

S: 221 Bye

{

"timestamp":"2016-1-15 12:00:00",

"from":"medcl@example.co",

"to":["lcdem@example.co"]

}

type smtpPrivateData struct{

	Data [2]*SmtpStream

}



type SmtpStream struct {

	tcptuple *common.TcpTuple



	data byte



	parseOffset int

	isClient    bool

	message *SmtpMessage

}



type SmtpMessage struct {

	Ts   time.Time

	From string

	To string

}

package smtp



import (

	"github.com/elastic/beats/libbeat/common"

	"github.com/elastic/beats/libbeat/logp"

	"github.com/elastic/beats/libbeat/publisher"

	"github.com/elastic/beats/packetbeat/config"

	"github.com/elastic/beats/packetbeat/protos"

	"github.com/elastic/beats/packetbeat/protos/tcp"

	"bytes"

	"time"

	"strings"

)



type smtpPrivateData struct{

	Data [2]*SmtpStream

}



type SmtpStream struct {

	tcptuple *common.TcpTuple



	data byte



	parseOffset int

	isClient    bool



	message *SmtpMessage

}



type SmtpMessage struct {

	start int

	end   int



	Ts   time.Time

	From string

	To string

	IgnoreMessage bool

}



type Smtp struct {

	SendRequest         bool

	SendResponse        bool

	transactionTimeout time.Duration

	Ports         int

	results publisher.Client

}



func (smtp *Smtp) initDefaults() {

	smtp.SendRequest = false

	smtp.SendResponse = false

	smtp.transactionTimeout = protos.DefaultTransactionExpiration

}



func (smtp *Smtp) setFromConfig(config config.Smtp) error {

	smtp.Ports = config.Ports

	if config.SendRequest != nil {

		smtp.SendRequest = *config.SendRequest

	}

	if config.SendResponse != nil {

		smtp.SendResponse = *config.SendResponse

	}



	if config.TransactionTimeout != nil && *config.TransactionTimeout > 0 {

		smtp.transactionTimeout = time.Duration(*config.TransactionTimeout) * time.Second

	}



	return nil

}



func (smtp *Smtp) GetPorts() int {

	return smtp.Ports

}



func (smtp *Smtp) Init(test_mode bool, results publisher.Client) error {

	smtp.initDefaults()



	if !test_mode {

		err := smtp.setFromConfig(config.ConfigSingleton.Protocols.Smtp)

		if err != nil {

			return err

		}

	}

	smtp.results = results



	return nil

}



func readLine(data byte, offset int) (bool, string, int) {

	q := bytes.Index(data[offset:], byte("\r\n"))

	if q == -1 {

		return false, "", 0

	}

	return true, string(data[offset : offset+q]), offset + q + 2

}



func (smtp *Smtp) Parse(pkt *protos.Packet, tcptuple *common.TcpTuple, dir uint8, private protos.ProtocolData, ) protos.ProtocolData {



	defer logp.Recover("ParseSmtp exception")



	priv := smtpPrivateData{}

	if private != nil {

		var ok bool

		priv, ok = private.(smtpPrivateData)

		if !ok {

			priv = smtpPrivateData{}

		}

	}



	if priv.Data[dir] == nil {

		priv.Data[dir] = &SmtpStream{

			tcptuple: tcptuple,

			data:     pkt.Payload,

			message:  &SmtpMessage{Ts: pkt.Ts},

		}

	} else {

		// concatenate bytes

		priv.Data[dir].data = append(priv.Data[dir].data, pkt.Payload...)

		if len(priv.Data[dir].data) > tcp.TCP_MAX_DATA_IN_STREAM {

			logp.Debug("smtp", "Stream data too large, dropping TCP stream")

			priv.Data[dir] = nil

			return priv

		}

	}



	stream := priv.Data[dir]

	for len(stream.data) > 0 {

		if stream.message == nil {

			stream.message = &SmtpMessage{Ts: pkt.Ts}

		}



		ok, complete := stmpMessageParser(priv.Data[dir])

		if !ok {

			// drop this tcp stream. Will retry parsing with the next

			// segment in it

			priv.Data[dir] = nil

			logp.Debug("smtp", "Ignore SMTP message. Drop tcp stream. Try parsing with the next segment")

			return priv

		}



		if complete {

			smtp.messageComplete(tcptuple, dir, stream)

		} else {

			logp.Debug("smtp","still wait message...")

			// wait for more data

			break

		}

	}



	return priv

}



func (smtp *Smtp) ConnectionTimeout() time.Duration {

	return smtp.transactionTimeout

}



func stmpMessageParser(s *SmtpStream) (bool, bool) {



	var value string=""



	for s.parseOffset < len(s.data) {





		logp.Debug("smtp", "Parse message: %s", string(s.data[s.parseOffset]))





		if strings.HasPrefix(string(s.data[s.parseOffset]),"MAIL" ) {



			logp.Debug("smtp", "Hit MAIL command: %s", string(s.data[s.parseOffset]))



			found, line, off := readLine(s.data, s.parseOffset)

			if !found {

				return true, false

			}



			value = line[1:]

			logp.Debug("smtp", "value  %s", value)



			s.parseOffset = off

		} else {

			logp.Debug("smtp", "Unexpected message starting with %s", s.data[s.parseOffset:])

			return false, false

		}

	}



	return true, false

}



func handleSmtp(stmp *Smtp, m *SmtpMessage, tcptuple *common.TcpTuple,

dir uint8, raw_msg byte) {

	logp.Info("smtp","handle smtp message...")



	//TODO



}



// Called when the parser has identified a full message.

func (smtp *Smtp) messageComplete(tcptuple *common.TcpTuple, dir uint8, stream *SmtpStream) {



	logp.Info("smtp","message completed...")



	// all ok, ship it

	msg := stream.data[stream.message.start:stream.message.end]



	if !stream.message.IgnoreMessage {

		handleSmtp(smtp, stream.message, tcptuple, dir, msg)

	}



	// and reset message

	stream.PrepareForNewMessage()

}



func (stream *SmtpStream) PrepareForNewMessage() {

	logp.Info("smtp","prepare for new message...")



	stream.data = stream.data[stream.parseOffset:]

	stream.parseOffset = 0

	stream.isClient = false

	stream.message = nil

}







func (smtp *Smtp) GapInStream(tcptuple *common.TcpTuple, dir uint8,

nbytes int, private protos.ProtocolData) (priv protos.ProtocolData, drop bool) {



	defer logp.Recover("GapInStream(smtp) exception")



	if private == nil {

		return private, false

	}



	return private, true

}



func (smtp *Smtp) ReceivedFin(tcptuple *common.TcpTuple, dir uint8,

private protos.ProtocolData) protos.ProtocolData {



	logp.Info("smtp","stream closed...")



	// TODO: check if we have data pending and either drop it to free

	// memory or send it up the stack.

	return private

}

cd ~/go/src/github.com/elastic/beats/packetbeat

make

./packetbeat -d "smtp" -c etc/packetbeat.yml -I ~/Downloads/smtp.pcap  -e -N

➜  packetbeat git:(smtpbeat) ✗ ./packetbeat -d "smtp" -c etc/packetbeat.yml -I ~/Downloads/smtp.pcap  -e -N 

2016/01/15 10:12:19.058535 publish.go:191: INFO Dry run mode. All output types except the file based one are disabled.

2016/01/15 10:12:19.058570 geolite.go:24: INFO GeoIP disabled: No paths were set under output.geoip.paths

2016/01/15 10:12:19.058592 publish.go:262: INFO Publisher name: medcls-MacBook.local

2016/01/15 10:12:19.058724 beat.go:145: INFO Init Beat: packetbeat; Version: 1.0.0

2016/01/15 10:12:19.059758 beat.go:171: INFO packetbeat sucessfully setup. Start running.

2016/01/15 10:12:20.155335 smtp.go:163: DBG  Parse message: 2

2016/01/15 10:12:20.155416 smtp.go:180: DBG  Unexpected message starting with 250-xc90.websitewelcome.com Hello GP [122.162.143.157]

250-SIZE 52428800

250-PIPELINING

250-AUTH PLAIN LOGIN

250-STARTTLS

250 HELP

2016/01/15 10:12:22.310974 smtp.go:163: DBG  Parse message: F

2016/01/15 10:12:22.311025 smtp.go:180: DBG  Unexpected message starting with From: "Gurpartap Singh" 

To: 

Subject: SMTP

Date: Mon, 5 Oct 2009 11:36:07 +0530

Message-ID: <000301ca4581$ef9e57f0$cedb07d0$@in>

MIME-Version: 1.0

...